feat: add Statement#stepIntoStrings() to read rows in batches

docs/api.md

@@ -431,6 +431,8 @@ An object representing a single SQL statement.
 - [Statement#get()](#getbindparameters---row)
 - [Statement#all()](#allbindparameters---array-of-rows)
 - [Statement#iterate()](#iteratebindparameters---iterator)
+- [Statement#stepIntoStrings()](#stepintostringsbuffer-capacity-bindparameters---array)
+- [Statement#closeCursor()](#closecursor---this)
 - [Statement#pluck()](#plucktogglestate---this)
 - [Statement#expand()](#expandtogglestate---this)
 - [Statement#raw()](#rawtogglestate---this)
@@ -513,6 +515,58 @@ for (const cat of stmt.iterate()) {
 }
 ```
 
+### .stepIntoStrings(*buffer*, *capacity*, [*...bindParameters*]) -> *array*
+
+**(only on statements that return data)*
+
+Reads up to `capacity` rows per call, for callers that consume many rows and want to avoid the cost of one native call and one row object per row. Numbers are written into `buffer`, an `ArrayBuffer` or `SharedArrayBuffer`. TEXT and BLOB values are created natively and returned in an array, one per call.
+
+`buffer` must hold at least `16 + 9 * capacity * columnCount` bytes, where `columnCount` is `stmt.columns().length`. A resizable `ArrayBuffer` is not accepted (a growable `SharedArrayBuffer` is). It is laid out as follows, with `R = capacity` and `C = columnCount`:
+
+| bytes | contents |
+|---|---|
+| `[0, 16)` | `Int32` × 4: the number of rows written, `done` (`1` if the result is exhausted, else `0`), `C`, and `0` |
+| `[16, 16 + 8RC)` | one 8-byte slot per value, row-major: an `Int64` for INTEGER, a `Float64` for FLOAT, and for TEXT and BLOB a `Uint32` index into the returned array (the other 4 bytes are `0`) |
+| `[16 + 8RC, 16 + 9RC)` | one tag byte per value: `0` NULL, `1` INTEGER, `2` FLOAT, `3` TEXT (a string), `4` BLOB (a `Buffer`) |
+
+Values are written as by [`.raw()`](#rawtogglestate---this): [`.pluck()`](#plucktogglestate---this), [`.expand()`](#expandtogglestate---this), [`.raw()`](#rawtogglestate---this) and [`.safeIntegers()`](./integer.md#getting-bigints-from-the-database) have no effect, and the caller decides how to convert integers.
+
+If a call does not exhaust the result, the statement stays busy like an open [`.iterate()`](#iteratebindparameters---iterator) iterator: call `.stepIntoStrings(buffer, capacity)` again, with no bind parameters, to continue, or [`.closeCursor()`](#closecursor---this) to stop early. A cursor that is neither finished nor closed keeps the statement busy, just like an iterator that is never finished or returned. If execution fails, an `Error` is thrown and the cursor is closed. A call can return `done` with no rows when the previous call returned exactly the last rows.
+
+```js
+const stmt = db.prepare('SELECT id, name, weight FROM cats');
+const capacity = 64;
+const columns = stmt.columns().length;
+const buffer = new ArrayBuffer(16 + 9 * capacity * columns);
+const header = new Int32Array(buffer, 0, 4);
+const f64 = new Float64Array(buffer, 16, capacity * columns);
+const i64 = new BigInt64Array(buffer, 16, capacity * columns);
+const u32 = new Uint32Array(buffer, 16, capacity * columns * 2);
+const tags = new Uint8Array(buffer, 16 + 8 * capacity * columns, capacity * columns);
+const value = (values, cell) => {
+  switch (tags[cell]) {
+    case 1: return i64[cell];
+    case 2: return f64[cell];
+    case 3: case 4: return values[u32[cell * 2]];
+    default: return null;
+  }
+};
+
+let values = stmt.stepIntoStrings(buffer, capacity);
+for (;;) {
+  for (let row = 0; row < header[0]; row++) {
+    const cell = row * columns;
+    console.log(value(values, cell), value(values, cell + 1), value(values, cell + 2));
+  }
+  if (header[1]) break;
+  values = stmt.stepIntoStrings(buffer, capacity);
+}
+```
+
+### .closeCursor() -> *this*
+
+Abandons a cursor started by [`.stepIntoStrings()`](#stepintostringsbuffer-capacity-bindparameters---array), resetting the statement so it can be used again. Does nothing if no cursor is open.
+
 ### .pluck([toggleState]) -> *this*
 
 **(only on statements that return data)*
@@ -621,7 +675,7 @@ console.log(cat.name); // => "Joey"
 
 **.readonly -> _boolean_** - Whether the prepared statement is readonly, meaning it does not mutate the database (note that [SQL functions might still change the database indirectly](https://www.sqlite.org/c3ref/stmt_readonly.html) as a side effect, even if the `.readonly` property is `true`).
 
-**.busy -> _boolean_** - Whether the prepared statement is busy executing a query via the [`.iterate()`](#iteratebindparameters---iterator) method.
+**.busy -> _boolean_** - Whether the prepared statement is busy executing a query via the [`.iterate()`](#iteratebindparameters---iterator) or [`.stepIntoStrings()`](#stepintostringsbuffer-capacity-bindparameters---array) method.
 
 # class *SqliteError*
 

lib/index.d.ts

@@ -28,6 +28,18 @@ declare namespace BetterSqlite3 {
         get(...params: BindParameters): Result | undefined;
         all(...params: BindParameters): Result[];
         iterate(...params: BindParameters): IterableIterator<Result>;
+        /**
+         * Steps up to `capacity` rows into `buffer` and returns the TEXT and BLOB
+         * values of those rows. See docs/api.md for the buffer layout. Bind
+         * parameters are only accepted by the call that starts a cursor.
+         */
+        stepIntoStrings(
+            buffer: ArrayBuffer | SharedArrayBuffer,
+            capacity: number,
+            ...params: BindParameters | []
+        ): Array<string | Buffer>;
+        /** Abandons a cursor started by stepIntoStrings(), if one is open. */
+        closeCursor(): this;
         pluck(toggleState?: boolean): this;
         expand(toggleState?: boolean): this;
         raw(toggleState?: boolean): this;

src/better_sqlite3.cpp

@@ -43,6 +43,7 @@ class Backup;
 
 #include "objects/backup.cpp"
 #include "objects/statement.cpp"
+#include "objects/statement-batch.cpp"
 #include "objects/database.cpp"
 #include "objects/statement-iterator.cpp"
 #include "util/unicode_case.cpp"

src/objects/statement-batch.cpp

@@ -0,0 +1,242 @@
+// Statement#stepIntoStrings(buffer, capacity, ...bindParameters) and
+// Statement#closeCursor(): read up to `capacity` rows per call.
+//
+// Numbers go into `buffer`, an ArrayBuffer or SharedArrayBuffer of at least
+// 16 + 9 * capacity * columnCount bytes, laid out as:
+//
+//   [0, 16)             int32 rowCount, int32 done (1 or 0), int32 columnCount,
+//                       int32 reserved (0)
+//   [16, +8*R*C)        one 8-byte slot per value, row-major:
+//                         INTEGER -> int64, FLOAT -> double,
+//                         TEXT/BLOB -> uint32 index into the returned array
+//   [16+8*R*C, +R*C)    one tag byte per value (see BatchTag)
+//
+// TEXT and BLOB values are created natively and returned as one array per call.
+//
+// Rows are produced as by raw(true); pluck/expand/raw and safeIntegers do not
+// apply, since integers are always written as int64.
+//
+// When a call does not exhaust the result (done is 0) the statement stays
+// locked like an open iterate() iterator, and the next stepIntoStrings(buffer,
+// capacity) call, with no bind parameters, continues it. closeCursor()
+// abandons it. As with an abandoned iterator, a cursor that is never finished
+// or closed keeps the statement locked and counts as an open iterator.
+
+namespace Batch {
+
+	enum BatchTag : uint8_t {
+		TAG_NULL = 0,
+		TAG_INT = 1,
+		TAG_FLOAT = 2,
+		TAG_TEXT = 3,
+		TAG_BLOB = 4,
+	};
+
+	static const size_t HEADER = 16;
+
+	struct Layout {
+		// Keeps the memory alive for the whole call: a user-defined function
+		// runs JavaScript during sqlite3_step(), which could detach the buffer
+		// (e.g. with transfer()) and so free its memory.
+		std::shared_ptr<v8::BackingStore> store;
+		uint8_t* slots;
+		uint8_t* tags;
+		int32_t* header;
+		int columns;
+
+		// Returns false if `buf` is not a (Shared)ArrayBuffer large enough.
+		// Resizable ArrayBuffers are rejected, since a user-defined function
+		// could shrink one while rows are being written.
+		bool Init(v8::Local<v8::Value> buf, int capacity, int columns_) {
+			size_t size;
+			if (buf->IsArrayBuffer()) {
+				v8::Local<v8::ArrayBuffer> ab = buf.As<v8::ArrayBuffer>();
+				store = ab->GetBackingStore();
+				if (store->IsResizableByUserJavaScript()) return false;
+				size = ab->ByteLength();
+			} else if (buf->IsSharedArrayBuffer()) {
+				// A growable SharedArrayBuffer only grows, in place.
+				v8::Local<v8::SharedArrayBuffer> sab = buf.As<v8::SharedArrayBuffer>();
+				store = sab->GetBackingStore();
+				size = sab->ByteLength();
+			} else {
+				return false;
+			}
+			void* data = store->Data();
+			const size_t cells = (size_t)capacity * (size_t)columns_;
+			if (data == NULL || size < HEADER + cells * 9) return false;
+			uint8_t* base = static_cast<uint8_t*>(data);
+			header = reinterpret_cast<int32_t*>(base);
+			slots = base + HEADER;
+			tags = slots + cells * 8;
+			columns = columns_;
+			return true;
+		}
+	};
+
+	static inline bool IsAscii(const uint8_t* p, size_t n) {
+		uint64_t acc = 0;
+		size_t i = 0;
+		for (; i + 8 <= n; i += 8) {
+			uint64_t w;
+			memcpy(&w, p + i, 8);
+			acc |= w;
+		}
+		for (; i < n; ++i) acc |= p[i];
+		return (acc & 0x8080808080808080ULL) == 0;
+	}
+
+	// Writes the current row at index `row`. Returns false, with a pending
+	// exception, if a value could not be converted.
+	static inline bool WriteRow(
+		v8::Isolate* isolate,
+		sqlite3_stmt* handle,
+		Layout& l,
+		int row,
+		v8::LocalVector<v8::Value>& values
+	) {
+		const int C = l.columns;
+		uint8_t* slot = l.slots + (size_t)row * C * 8;
+		uint8_t* tag = l.tags + (size_t)row * C;
+		for (int i = 0; i < C; ++i, slot += 8) {
+			switch (sqlite3_column_type(handle, i)) {
+				case SQLITE_INTEGER: {
+					sqlite3_int64 v = sqlite3_column_int64(handle, i);
+					memcpy(slot, &v, 8);
+					tag[i] = TAG_INT;
+					break;
+				}
+				case SQLITE_FLOAT: {
+					double v = sqlite3_column_double(handle, i);
+					memcpy(slot, &v, 8);
+					tag[i] = TAG_FLOAT;
+					break;
+				}
+				case SQLITE_TEXT: {
+					const uint8_t* p = sqlite3_column_text(handle, i);
+					int n = sqlite3_column_bytes(handle, i);
+					v8::MaybeLocal<v8::String> s = IsAscii(p, (size_t)n)
+						? v8::String::NewFromOneByte(isolate, p, v8::NewStringType::kNormal, n)
+						: v8::String::NewFromUtf8(isolate, reinterpret_cast<const char*>(p), v8::NewStringType::kNormal, n);
+					if (s.IsEmpty()) {
+						ThrowRangeError("The string is too long to be returned to JavaScript");
+						return false;
+					}
+					uint32_t index[2] = {(uint32_t)values.size(), 0};
+					memcpy(slot, index, 8);
+					tag[i] = TAG_TEXT;
+					values.emplace_back(s.ToLocalChecked());
+					break;
+				}
+				case SQLITE_BLOB: {
+					v8::MaybeLocal<v8::Object> b = node::Buffer::Copy(
+						isolate,
+						static_cast<const char*>(sqlite3_column_blob(handle, i)),
+						sqlite3_column_bytes(handle, i)
+					);
+					if (b.IsEmpty()) return false;
+					uint32_t index[2] = {(uint32_t)values.size(), 0};
+					memcpy(slot, index, 8);
+					tag[i] = TAG_BLOB;
+					values.emplace_back(b.ToLocalChecked());
+					break;
+				}
+				default:
+					memset(slot, 0, 8);
+					tag[i] = TAG_NULL;
+					break;
+			}
+		}
+		return true;
+	}
+
+}
+
+// Ends an open cursor: resets the statement and releases its lock. Safe to call
+// after the database is closed (the handle is then already finalized).
+void Statement::CloseCursor() {
+	if (!cursor_open) return;
+	cursor_open = false;
+	locked = false;
+	db->GetState()->iterators -= 1;
+	if (alive) {
+		sqlite3_reset(handle);
+		if (!bound) sqlite3_clear_bindings(handle);
+	}
+}
+
+NODE_METHOD(Statement::JS_stepIntoStrings) {
+	Statement* stmt = Unwrap<Statement>(info.This());
+	REQUIRE_STATEMENT_RETURNS_DATA();
+	sqlite3_stmt* handle = stmt->handle;
+	Database* db = stmt->db;
+	REQUIRE_DATABASE_OPEN(db->GetState());
+	REQUIRE_DATABASE_NOT_BUSY(db->GetState());
+	if (info.Length() < 2 || !info[1]->IsInt32() || info[1].As<v8::Int32>()->Value() < 1) {
+		return ThrowTypeError("Expected second argument to be a positive 32-bit integer");
+	}
+	const int capacity = info[1].As<v8::Int32>()->Value();
+	Batch::Layout l;
+	if (!l.Init(info[0], capacity, sqlite3_column_count(handle))) {
+		return ThrowRangeError("Expected first argument to be an ArrayBuffer or SharedArrayBuffer of at least 16 + 9 * capacity * columnCount bytes");
+	}
+	UseIsolate;
+
+	if (!stmt->cursor_open) {
+		REQUIRE_STATEMENT_NOT_LOCKED(stmt);
+		if (db->GetState()->iterators == USHRT_MAX) return ThrowRangeError("Too many active database iterators");
+		if (!stmt->bound) {
+			Binder binder(handle);
+			if (!binder.BindFrom(info, 2, stmt)) {
+				sqlite3_clear_bindings(handle);
+				return;
+			}
+		} else if (info.Length() > 2) {
+			return ThrowTypeError("This statement already has bound parameters");
+		}
+		if (db->Log(isolate, handle)) {
+			if (!stmt->bound) sqlite3_clear_bindings(handle);
+			return db->ThrowDatabaseError();
+		}
+		stmt->cursor_open = true;
+		stmt->locked = true;
+		db->GetState()->iterators += 1;
+	} else if (info.Length() > 2) {
+		return ThrowTypeError("Bind parameters can only be given to the stepIntoStrings() call that starts a cursor");
+	}
+
+	v8::LocalVector<v8::Value> values(isolate);
+	values.reserve((size_t)std::min(capacity, 64) * (size_t)l.columns);
+	int rows = 0;
+	int status = SQLITE_ROW;
+	bool converted = true;
+	db->GetState()->busy = true;
+	while (rows < capacity && (status = sqlite3_step(handle)) == SQLITE_ROW) {
+		if (!(converted = Batch::WriteRow(isolate, handle, l, rows, values))) break;
+		++rows;
+	}
+	db->GetState()->busy = false;
+
+	if (!converted) {
+		stmt->CloseCursor();
+		return;
+	}
+	if (status != SQLITE_ROW && status != SQLITE_DONE) {
+		stmt->CloseCursor();
+		return db->ThrowDatabaseError();
+	}
+	const bool done = status == SQLITE_DONE;
+	if (done) stmt->CloseCursor();
+	l.header[0] = rows;
+	l.header[1] = done ? 1 : 0;
+	l.header[2] = l.columns;
+	l.header[3] = 0;
+	info.GetReturnValue().Set(v8::Array::New(isolate, values.data(), values.size()));
+}
+
+NODE_METHOD(Statement::JS_closeCursor) {
+	Statement* stmt = Unwrap<Statement>(info.This());
+	REQUIRE_DATABASE_NOT_BUSY(stmt->db->GetState());
+	stmt->CloseCursor();
+	info.GetReturnValue().Set(info.This());
+}

src/objects/statement.cpp

@@ -14,7 +14,8 @@ Statement::Statement(
 	has_bind_map(false),
 	safe_ints(db->GetState()->safe_ints),
 	mode(Data::FLAT),
-	returns_data(returns_data) {
+	returns_data(returns_data),
+	cursor_open(false) {
 	assert(db != NULL);
 	assert(handle != NULL);
 	assert(db->GetState()->open);
@@ -82,6 +83,8 @@ INIT(Statement::Init) {
 	SetPrototypeMethod(isolate, data, t, "scanStatusV2", JS_scanStatusV2);
 	SetPrototypeMethod(isolate, data, t, "scanStatusReset", JS_scanStatusReset);
 	SetPrototypeMethod(isolate, data, t, "explainQueryPlan", JS_explainQueryPlan);
+	SetPrototypeMethod(isolate, data, t, "stepIntoStrings", JS_stepIntoStrings);
+	SetPrototypeMethod(isolate, data, t, "closeCursor", JS_closeCursor);
 	SetPrototypeGetter(isolate, data, t, "busy", JS_busy);
 	return t->GetFunction(OnlyContext).ToLocalChecked();
 }

src/objects/statement.hpp

@@ -14,6 +14,9 @@ class Statement : public node::ObjectWrap { friend class StatementIterator;
 	// Returns the Statement's bind map (creates it upon first execution).
 	BindMap* GetBindMap(v8::Isolate* isolate);
 
+	// Ends a cursor opened by stepIntoStrings(), if any.
+	void CloseCursor();
+
 #if defined(NODE_MODULE_VERSION) && NODE_MODULE_VERSION >= 127
 	// Returns a builder of object rows that reuses the Statement's cached
 	// column names.
@@ -60,6 +63,8 @@ class Statement : public node::ObjectWrap { friend class StatementIterator;
 	static NODE_METHOD(JS_scanStatusReset);
 	static NODE_METHOD(JS_explainQueryPlan);
 	static NODE_GETTER(JS_busy);
+	static NODE_METHOD(JS_stepIntoStrings);
+	static NODE_METHOD(JS_closeCursor);
 
 	Database* const db;
 	sqlite3_stmt* const handle;
@@ -71,4 +76,5 @@ class Statement : public node::ObjectWrap { friend class StatementIterator;
 	bool safe_ints;
 	char mode;
 	const bool returns_data;
+	bool cursor_open;
 };