Repository: nodejs/node
Pull request: sqlite: add aggregate function

Closes #56511

inspired by https://github.com/WiseLibs/better-sqlite3/blob/master/docs/api.md#aggregatename-options---this

Changed files:
filename: doc/api/sqlite.md
patch: @@ -119,6 +119,85 @@ added: v22.5.0
 
 Constructs a new `DatabaseSync` instance.
 
+### `database.aggregate(name, options)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+Registers a new aggregate function with the SQLite database. This method is a wrapper around
+[`sqlite3_create_window_function()`][].
+
+* `name` {string} The name of the SQLite function to create.
+* `options` {Object} Function configuration settings.
+  * `deterministic` {boolean} If `true`, the [`SQLITE_DETERMINISTIC`][] flag is
+    set on the created function. **Default:** `false`.
+  * `directOnly` {boolean} If `true`, the [`SQLITE_DIRECTONLY`][] flag is set on
+    the created function. **Default:** `false`.
+  * `useBigIntArguments` {boolean} If `true`, integer arguments to `options.step` and `options.inverse`
+    are converted to `BigInt`s. If `false`, integer arguments are passed as
+    JavaScript numbers. **Default:** `false`.
+  * `varargs` {boolean} If `true`, `options.step` and `options.inverse` may be invoked with any number of
+    arguments (between zero and [`SQLITE_MAX_FUNCTION_ARG`][]). If `false`,
+    `inverse` and `step` must be invoked with exactly `length` arguments.
+    **Default:** `false`.
+  * `start` {number | string | null | Array | Object | Function} The identity
+    value for the aggregation function. This value is used when the aggregation
+    function is initialized. When a {Function} is passed the identity will be its return value.
+  * `step` {Function} The function to call for each row in the aggregation. The
+    function receives the current state and the row value. The return value of
+    this function should be the new state.
+  * `result` {Function} The function to call to get the result of the
+    aggregation. The function receives the final state and should return the
+    result of the aggregation.
+  * `inverse` {Function} When this function is provided, the `aggregate` method will work as a window function.
+    The function receives the current state and the dropped row value. The return value of this function should be the
+    new state.
+
+When used as a window function, the `result` function will be called multiple times.
+
+```cjs
+const { DatabaseSync } = require('node:sqlite');
+
+const db = new DatabaseSync(':memory:');
+db.exec(`
+  CREATE TABLE t3(x, y);
+  INSERT INTO t3 VALUES ('a', 4),
+                        ('b', 5),
+                        ('c', 3),
+                        ('d', 8),
+                        ('e', 1);
+`);
+
+db.aggregate('sumint', {
+  start: 0,
+  step: (acc, value) => acc + value,
+});
+
+db.prepare('SELECT sumint(y) as total FROM t3').get(); // { total: 21 }
+```
+
+```mjs
+import { DatabaseSync } from 'node:sqlite';
+
+const db = new DatabaseSync(':memory:');
+db.exec(`
+  CREATE TABLE t3(x, y);
+  INSERT INTO t3 VALUES ('a', 4),
+                        ('b', 5),
+                        ('c', 3),
+                        ('d', 8),
+                        ('e', 1);
+`);
+
+db.aggregate('sumint', {
+  start: 0,
+  step: (acc, value) => acc + value,
+});
+
+db.prepare('SELECT sumint(y) as total FROM t3').get(); // { total: 21 }
+```
+
 ### `database.close()`
 
 <!-- YAML
@@ -751,6 +830,7 @@ resolution handler passed to [`database.applyChangeset()`][]. See also
 [`sqlite3_column_origin_name()`]: https://www.sqlite.org/c3ref/column_database_name.html
 [`sqlite3_column_table_name()`]: https://www.sqlite.org/c3ref/column_database_name.html
 [`sqlite3_create_function_v2()`]: https://www.sqlite.org/c3ref/create_function.html
+[`sqlite3_create_window_function()`]: https://www.sqlite.org/c3ref/create_function.html
 [`sqlite3_exec()`]: https://www.sqlite.org/c3ref/exec.html
 [`sqlite3_expanded_sql()`]: https://www.sqlite.org/c3ref/expanded_sql.html
 [`sqlite3_last_insert_rowid()`]: https://www.sqlite.org/c3ref/last_insert_rowid.html

filename: src/env_properties.h
patch: @@ -207,6 +207,7 @@
   V(inputs_string, "inputs")                                                   \
   V(internal_binding_string, "internalBinding")                                \
   V(internal_string, "internal")                                               \
+  V(inverse_string, "inverse")                                                 \
   V(ipv4_string, "IPv4")                                                       \
   V(ipv6_string, "IPv6")                                                       \
   V(isclosing_string, "isClosing")                                             \
@@ -334,6 +335,7 @@
     "const __esModule = true;")                                                \
   V(require_string, "require")                                                 \
   V(resource_string, "resource")                                               \
+  V(result_string, "result")                                                   \
   V(retry_string, "retry")                                                     \
   V(return_arrays_string, "returnArrays")                                      \
   V(return_string, "return")                                                   \
@@ -361,12 +363,14 @@
   V(specifier_string, "specifier")                                             \
   V(stack_string, "stack")                                                     \
   V(standard_name_string, "standardName")                                      \
+  V(start_string, "start")                                                     \
   V(start_time_string, "startTime")                                            \
   V(state_string, "state")                                                     \
   V(statement_string, "statement")                                             \
   V(stats_string, "stats")                                                     \
   V(status_string, "status")                                                   \
   V(stdio_string, "stdio")                                                     \
+  V(step_string, "step")                                                       \
   V(stream_average_duration_string, "streamAverageDuration")                   \
   V(stream_count_string, "streamCount")                                        \
   V(subject_string, "subject")                                                 \

filename: src/node_sqlite.cc
patch: @@ -169,6 +169,38 @@ inline MaybeLocal<Object> CreateSQLiteError(Isolate* isolate, sqlite3* db) {
   return e;
 }
 
+void JSValueToSQLiteResult(Isolate* isolate,
+                           sqlite3_context* ctx,
+                           Local<Value> value) {
+  if (value->IsNullOrUndefined()) {
+    sqlite3_result_null(ctx);
+  } else if (value->IsNumber()) {
+    sqlite3_result_double(ctx, value.As<Number>()->Value());
+  } else if (value->IsString()) {
+    Utf8Value val(isolate, value.As<String>());
+    sqlite3_result_text(ctx, *val, val.length(), SQLITE_TRANSIENT);
+  } else if (value->IsArrayBufferView()) {
+    ArrayBufferViewContents<uint8_t> buf(value);
+    sqlite3_result_blob(ctx, buf.data(), buf.length(), SQLITE_TRANSIENT);
+  } else if (value->IsBigInt()) {
+    bool lossless;
+    int64_t as_int = value.As<BigInt>()->Int64Value(&lossless);
+    if (!lossless) {
+      sqlite3_result_error(ctx, "BigInt value is too large for SQLite", -1);
+      return;
+    }
+    sqlite3_result_int64(ctx, as_int);
+  } else if (value->IsPromise()) {
+    sqlite3_result_error(
+        ctx, "Asynchronous user-defined functions are not supported", -1);
+  } else {
+    sqlite3_result_error(
+        ctx,
+        "Returned JavaScript value cannot be converted to a SQLite value",
+        -1);
+  }
+}
+
 class DatabaseSync;
 
 inline void THROW_ERR_SQLITE_ERROR(Isolate* isolate, DatabaseSync* db) {
@@ -215,6 +247,178 @@ inline MaybeLocal<Value> NullableSQLiteStringToValue(Isolate* isolate,
       .As<Value>();
 }
 
+class CustomAggregate {
+ public:
+  explicit CustomAggregate(Environment* env,
+                           DatabaseSync* db,
+                           bool use_bigint_args,
+                           Local<Value> start,
+                           Local<Function> step_fn,
+                           Local<Function> inverse_fn,
+                           Local<Function> result_fn)
+      : env_(env),
+        db_(db),
+        use_bigint_args_(use_bigint_args),
+        start_(env->isolate(), start),
+        step_fn_(env->isolate(), step_fn),
+        inverse_fn_(env->isolate(), inverse_fn),
+        result_fn_(env->isolate(), result_fn) {}
+
+  static void xStep(sqlite3_context* ctx, int argc, sqlite3_value** argv) {
+    xStepBase(ctx, argc, argv, &CustomAggregate::step_fn_);
+  }
+
+  static void xInverse(sqlite3_context* ctx, int argc, sqlite3_value** argv) {
+    xStepBase(ctx, argc, argv, &CustomAggregate::inverse_fn_);
+  }
+
+  static void xFinal(sqlite3_context* ctx) { xValueBase(ctx, true); }
+
+  static void xValue(sqlite3_context* ctx) { xValueBase(ctx, false); }
+
+  static void xDestroy(void* self) {
+    delete static_cast<CustomAggregate*>(self);
+  }
+
+ private:
+  struct aggregate_data {
+    Global<Value> value;
+    bool initialized;
+    bool is_window;
+  };
+
+  static inline void xStepBase(sqlite3_context* ctx,
+                               int argc,
+                               sqlite3_value** argv,
+                               Global<Function> CustomAggregate::*mptr) {
+    CustomAggregate* self =
+        static_cast<CustomAggregate*>(sqlite3_user_data(ctx));
+    Environment* env = self->env_;
+    Isolate* isolate = env->isolate();
+    auto agg = self->GetAggregate(ctx);
+
+    if (!agg) {
+      return;
+    }
+
+    auto recv = Undefined(isolate);
+    LocalVector<Value> js_argv(isolate);
+    js_argv.emplace_back(Local<Value>::New(isolate, agg->value));
+
+    for (int i = 0; i < argc; ++i) {
+      sqlite3_value* value = argv[i];
+      MaybeLocal<Value> js_val;
+      SQLITE_VALUE_TO_JS(value, isolate, self->use_bigint_args_, js_val, value);
+      if (js_val.IsEmpty()) {
+        // Ignore the SQLite error because a JavaScript exception is pending.
+        self->db_->SetIgnoreNextSQLiteError(true);
+        sqlite3_result_error(ctx, "", 0);
+        return;
+      }
+
+      Local<Value> local;
+      if (!js_val.ToLocal(&local)) {
+        // Ignore the SQLite error because a JavaScript exception is pending.
+        self->db_->SetIgnoreNextSQLiteError(true);
+        sqlite3_result_error(ctx, "", 0);
+        return;
+      }
+
+      js_argv.emplace_back(local);
+    }
+
+    Local<Value> ret;
+    if (!(self->*mptr)
+             .Get(isolate)
+             ->Call(env->context(), recv, argc + 1, js_argv.data())
+             .ToLocal(&ret)) {
+      self->db_->SetIgnoreNextSQLiteError(true);
+      sqlite3_result_error(ctx, "", 0);
+      return;
+    }
+
+    agg->value.Reset(isolate, ret);
+  }
+
+  static inline void xValueBase(sqlite3_context* ctx, bool is_final) {
+    CustomAggregate* self =
+        static_cast<CustomAggregate*>(sqlite3_user_data(ctx));
+    Environment* env = self->env_;
+    Isolate* isolate = env->isolate();
+    auto agg = self->GetAggregate(ctx);
+
+    if (!agg) {
+      return;
+    }
+
+    if (!is_final) {
+      agg->is_window = true;
+    } else if (agg->is_window) {
+      DestroyAggregateData(ctx);
+      return;
+    }
+
+    Local<Value> result;
+    if (!self->result_fn_.IsEmpty()) {
+      Local<Function> fn =
+          Local<Function>::New(env->isolate(), self->result_fn_);
+      Local<Value> js_arg[] = {Local<Value>::New(isolate, agg->value)};
+
+      if (!fn->Call(env->context(), Null(isolate), 1, js_arg)
+               .ToLocal(&result)) {
+        self->db_->SetIgnoreNextSQLiteError(true);
+        sqlite3_result_error(ctx, "", 0);
+      }
+    } else {
+      result = Local<Value>::New(isolate, agg->value);
+    }
+
+    JSValueToSQLiteResult(isolate, ctx, result);
+    if (is_final) {
+      DestroyAggregateData(ctx);
+    }
+  }
+
+  static void DestroyAggregateData(sqlite3_context* ctx) {
+    aggregate_data* agg = static_cast<aggregate_data*>(
+        sqlite3_aggregate_context(ctx, sizeof(aggregate_data)));
+    CHECK(agg->initialized);
+    agg->value.Reset();
+  }
+
+  aggregate_data* GetAggregate(sqlite3_context* ctx) {
+    aggregate_data* agg = static_cast<aggregate_data*>(
+        sqlite3_aggregate_context(ctx, sizeof(aggregate_data)));
+    if (!agg->initialized) {
+      Isolate* isolate = env_->isolate();
+      Local<Value> start_v = Local<Value>::New(isolate, start_);
+      if (start_v->IsFunction()) {
+        auto fn = start_v.As<Function>();
+        MaybeLocal<Value> retval =
+            fn->Call(env_->context(), Null(isolate), 0, nullptr);
+        if (!retval.ToLocal(&start_v)) {
+          db_->SetIgnoreNextSQLiteError(true);
+          sqlite3_result_error(ctx, "", 0);
+          return nullptr;
+        }
+      }
+
+      agg->value.Reset(env_->isolate(), start_v);
+      agg->initialized = true;
+    }
+
+    return agg;
+  }
+
+  Environment* env_;
+  DatabaseSync* db_;
+  bool use_bigint_args_;
+  Global<Value> start_;
+  Global<Function> step_fn_;
+  Global<Function> inverse_fn_;
+  Global<Function> result_fn_;
+};
+
 class BackupJob : public ThreadPoolWork {
  public:
   explicit BackupJob(Environment* env,
@@ -433,33 +637,7 @@ void UserDefinedFunction::xFunc(sqlite3_context* ctx,
     return;
   }
 
-  if (result->IsUndefined() || result->IsNull()) {
-    sqlite3_result_null(ctx);
-  } else if (result->IsNumber()) {
-    sqlite3_result_double(ctx, result.As<Number>()->Value());
-  } else if (result->IsString()) {
-    Utf8Value val(isolate, result.As<String>());
-    sqlite3_result_text(ctx, *val, val.length(), SQLITE_TRANSIENT);
-  } else if (result->IsArrayBufferView()) {
-    ArrayBufferViewContents<uint8_t> buf(result);
-    sqlite3_result_blob(ctx, buf.data(), buf.length(), SQLITE_TRANSIENT);
-  } else if (result->IsBigInt()) {
-    bool lossless;
-    int64_t as_int = result.As<BigInt>()->Int64Value(&lossless);
-    if (!lossless) {
-      sqlite3_result_error(ctx, "BigInt value is too large for SQLite", -1);
-      return;
-    }
-    sqlite3_result_int64(ctx, as_int);
-  } else if (result->IsPromise()) {
-    sqlite3_result_error(
-        ctx, "Asynchronous user-defined functions are not supported", -1);
-  } else {
-    sqlite3_result_error(
-        ctx,
-        "Returned JavaScript value cannot be converted to a SQLite value",
-        -1);
-  }
+  JSValueToSQLiteResult(isolate, ctx, result);
 }
 
 void UserDefinedFunction::xDestroy(void* self) {
@@ -978,6 +1156,163 @@ void DatabaseSync::CustomFunction(const FunctionCallbackInfo<Value>& args) {
   CHECK_ERROR_OR_THROW(env->isolate(), db, r, SQLITE_OK, void());
 }
 
+void DatabaseSync::AggregateFunction(const FunctionCallbackInfo<Value>& args) {
+  DatabaseSync* db;
+  ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
+  Environment* env = Environment::GetCurrent(args);
+  THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  Utf8Value name(env->isolate(), args[0].As<String>());
+  Local<Object> options = args[1].As<Object>();
+  Local<Value> start_v;
+  if (!options->Get(env->context(), env->start_string()).ToLocal(&start_v)) {
+    return;
+  }
+
+  if (start_v->IsUndefined()) {
+    THROW_ERR_INVALID_ARG_TYPE(env->isolate(),
+                               "The \"options.start\" argument must be a "
+                               "function or a primitive value.");
+    return;
+  }
+
+  Local<Value> step_v;
+  if (!options->Get(env->context(), env->step_string()).ToLocal(&step_v)) {
+    return;
+  }
+
+  if (!step_v->IsFunction()) {
+    THROW_ERR_INVALID_ARG_TYPE(
+        env->isolate(), "The \"options.step\" argument must be a function.");
+    return;
+  }
+
+  Local<Value> result_v;
+  if (!options->Get(env->context(), env->result_string()).ToLocal(&result_v)) {
+    return;
+  }
+
+  bool use_bigint_args = false;
+  bool varargs = false;
+  bool direct_only = false;
+  Local<Value> use_bigint_args_v;
+  Local<Function> inverseFunc = Local<Function>();
+  if (!options
+           ->Get(env->context(),
+                 FIXED_ONE_BYTE_STRING(env->isolate(), "useBigIntArguments"))
+           .ToLocal(&use_bigint_args_v)) {
+    return;
+  }
+
+  if (!use_bigint_args_v->IsUndefined()) {
+    if (!use_bigint_args_v->IsBoolean()) {
+      THROW_ERR_INVALID_ARG_TYPE(
+          env->isolate(),
+          "The \"options.useBigIntArguments\" argument must be a boolean.");
+      return;
+    }
+    use_bigint_args = use_bigint_args_v.As<Boolean>()->Value();
+  }
+
+  Local<Value> varargs_v;
+  if (!options
+           ->Get(env->context(),
+                 FIXED_ONE_BYTE_STRING(env->isolate(), "varargs"))
+           .ToLocal(&varargs_v)) {
+    return;
+  }
+
+  if (!varargs_v->IsUndefined()) {
+    if (!varargs_v->IsBoolean()) {
+      THROW_ERR_INVALID_ARG_TYPE(
+          env->isolate(),
+          "The \"options.varargs\" argument must be a boolean.");
+      return;
+    }
+    varargs = varargs_v.As<Boolean>()->Value();
+  }
+
+  Local<Value> direct_only_v;
+  if (!options
+           ->Get(env->context(),
+                 FIXED_ONE_BYTE_STRING(env->isolate(), "directOnly"))
+           .ToLocal(&direct_only_v)) {
+    return;
+  }
+
+  if (!direct_only_v->IsUndefined()) {
+    if (!direct_only_v->IsBoolean()) {
+      THROW_ERR_INVALID_ARG_TYPE(
+          env->isolate(),
+          "The \"options.directOnly\" argument must be a boolean.");
+      return;
+    }
+    direct_only = direct_only_v.As<Boolean>()->Value();
+  }
+
+  Local<Value> inverse_v;
+  if (!options->Get(env->context(), env->inverse_string())
+           .ToLocal(&inverse_v)) {
+    return;
+  }
+
+  if (!inverse_v->IsUndefined()) {
+    if (!inverse_v->IsFunction()) {
+      THROW_ERR_INVALID_ARG_TYPE(
+          env->isolate(),
+          "The \"options.inverse\" argument must be a function.");
+      return;
+    }
+    inverseFunc = inverse_v.As<Function>();
+  }
+
+  Local<Function> stepFunction = step_v.As<Function>();
+  Local<Function> resultFunction =
+      result_v->IsFunction() ? result_v.As<Function>() : Local<Function>();
+  int argc = -1;
+  if (!varargs) {
+    Local<Value> js_len;
+    if (!stepFunction->Get(env->context(), env->length_string())
+             .ToLocal(&js_len)) {
+      return;
+    }
+
+    // Subract 1 because the first argument is the aggregate value.
+    argc = js_len.As<Int32>()->Value() - 1;
+    if (!inverseFunc.IsEmpty() &&
+        !inverseFunc->Get(env->context(), env->length_string())
+             .ToLocal(&js_len)) {
+      return;
+    }
+
+    argc = std::max({argc, js_len.As<Int32>()->Value() - 1, 0});
+  }
+
+  int text_rep = SQLITE_UTF8;
+  if (direct_only) {
+    text_rep |= SQLITE_DIRECTONLY;
+  }
+
+  auto xInverse = !inverseFunc.IsEmpty() ? CustomAggregate::xInverse : nullptr;
+  auto xValue = xInverse ? CustomAggregate::xValue : nullptr;
+  int r = sqlite3_create_window_function(db->connection_,
+                                         *name,
+                                         argc,
+                                         text_rep,
+                                         new CustomAggregate(env,
+                                                             db,
+                                                             use_bigint_args,
+                                                             start_v,
+                                                             stepFunction,
+                                                             inverseFunc,
+                                                             resultFunction),
+                                         CustomAggregate::xStep,
+                                         CustomAggregate::xFinal,
+                                         xValue,
+                                         xInverse,
+                                         CustomAggregate::xDestroy);
+  CHECK_ERROR_OR_THROW(env->isolate(), db, r, SQLITE_OK, void());
+}
+
 void DatabaseSync::CreateSession(const FunctionCallbackInfo<Value>& args) {
   std::string table;
   std::string db_name = "main";
@@ -2253,6 +2588,8 @@ static void Initialize(Local<Object> target,
   SetProtoMethod(isolate, db_tmpl, "prepare", DatabaseSync::Prepare);
   SetProtoMethod(isolate, db_tmpl, "exec", DatabaseSync::Exec);
   SetProtoMethod(isolate, db_tmpl, "function", DatabaseSync::CustomFunction);
+  SetProtoMethod(
+      isolate, db_tmpl, "aggregate", DatabaseSync::AggregateFunction);
   SetProtoMethod(
       isolate, db_tmpl, "createSession", DatabaseSync::CreateSession);
   SetProtoMethod(

filename: src/node_sqlite.h
patch: @@ -60,6 +60,8 @@ class DatabaseSync : public BaseObject {
   static void Prepare(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Exec(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void CustomFunction(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void AggregateFunction(
+      const v8::FunctionCallbackInfo<v8::Value>& args);
   static void CreateSession(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void ApplyChangeset(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void EnableLoadExtension(

filename: test/parallel/test-sqlite-aggregate-function.mjs
patch: @@ -0,0 +1,393 @@
+import '../common/index.mjs';
+import { DatabaseSync } from 'node:sqlite';
+import { describe, test } from 'node:test';
+
+describe('DatabaseSync.prototype.aggregate()', () => {
+  describe('input validation', () => {
+    const db = new DatabaseSync(':memory:');
+
+    test('throws if options.start is not provided', (t) => {
+      t.assert.throws(() => {
+        db.aggregate('sum', {
+          result: (total) => total
+        });
+      }, {
+        code: 'ERR_INVALID_ARG_TYPE',
+        message: 'The "options.start" argument must be a function or a primitive value.'
+      });
+    });
+
+    test('throws if options.step is not a function', (t) => {
+      t.assert.throws(() => {
+        db.aggregate('sum', {
+          start: 0,
+          result: (total) => total
+        });
+      }, {
+        code: 'ERR_INVALID_ARG_TYPE',
+        message: 'The "options.step" argument must be a function.'
+      });
+    });
+
+    test('throws if options.useBigIntArguments is not a boolean', (t) => {
+      t.assert.throws(() => {
+        db.aggregate('sum', {
+          start: 0,
+          step: () => null,
+          useBigIntArguments: ''
+        });
+      }, {
+        code: 'ERR_INVALID_ARG_TYPE',
+        message: /The "options\.useBigIntArguments" argument must be a boolean/,
+      });
+    });
+
+    test('throws if options.varargs is not a boolean', (t) => {
+      t.assert.throws(() => {
+        db.aggregate('sum', {
+          start: 0,
+          step: () => null,
+          varargs: ''
+        });
+      }, {
+        code: 'ERR_INVALID_ARG_TYPE',
+        message: /The "options\.varargs" argument must be a boolean/,
+      });
+    });
+
+    test('throws if options.directOnly is not a boolean', (t) => {
+      t.assert.throws(() => {
+        db.aggregate('sum', {
+          start: 0,
+          step: () => null,
+          directOnly: ''
+        });
+      }, {
+        code: 'ERR_INVALID_ARG_TYPE',
+        message: /The "options\.directOnly" argument must be a boolean/,
+      });
+    });
+  });
+});
+
+describe('varargs', () => {
+  test('supports variable number of arguments when true', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.exec('CREATE TABLE data (value INTEGER)');
+    db.exec('INSERT INTO data VALUES (1), (2), (3)');
+    db.aggregate('sum_int', {
+      start: 0,
+      step: (_acc, _value, var1, var2, var3) => {
+        return var1 + var2 + var3;
+      },
+      varargs: true,
+    });
+
+    const result = db.prepare('SELECT sum_int(value, 1, 2, 3) as total FROM data').get();
+
+    t.assert.deepStrictEqual(result, { __proto__: null, total: 6 });
+  });
+
+  test('uses the max between step.length and inverse.length when false', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.exec(`
+      CREATE TABLE t3(x, y);
+      INSERT INTO t3 VALUES ('a', 1),
+                            ('b', 2),
+                            ('c', 3);
+    `);
+
+    db.aggregate('sumint', {
+      start: 0,
+      step: (acc, var1) => {
+        return var1 + acc;
+      },
+      inverse: (acc, var1, var2) => {
+        return acc - var1 - var2;
+      },
+      varargs: false,
+    });
+
+    const result = db.prepare(`
+      SELECT x, sumint(y, 10) OVER (
+        ORDER BY x ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING
+      ) AS sum_y
+      FROM t3 ORDER BY x;
+    `).all();
+
+    t.assert.deepStrictEqual(result, [
+      { __proto__: null, x: 'a', sum_y: 3 },
+      { __proto__: null, x: 'b', sum_y: 6 },
+      { __proto__: null, x: 'c', sum_y: -5 },
+    ]);
+
+    t.assert.throws(() => {
+      db.prepare(`
+        SELECT x, sumint(y) OVER (
+          ORDER BY x ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING
+        ) AS sum_y
+        FROM t3 ORDER BY x;
+      `);
+    }, {
+      code: 'ERR_SQLITE_ERROR',
+      message: 'wrong number of arguments to function sumint()'
+    });
+  });
+
+  test('throws if an incorrect number of arguments is provided when false', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.aggregate('sum_int', {
+      start: 0,
+      step: (_acc, var1, var2, var3) => {
+        return var1 + var2 + var3;
+      },
+      varargs: false,
+    });
+
+    t.assert.throws(() => {
+      db.prepare('SELECT sum_int(1, 2, 3, 4)').get();
+    }, {
+      code: 'ERR_SQLITE_ERROR',
+      message: 'wrong number of arguments to function sum_int()'
+    });
+  });
+});
+
+describe('directOnly', () => {
+  test('is false by default', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.aggregate('func', {
+      start: 0,
+      step: (acc, value) => acc + value,
+      inverse: (acc, value) => acc - value,
+    });
+    db.exec(`
+      CREATE TABLE t3(x, y);
+      INSERT INTO t3 VALUES ('a', 4),
+                            ('b', 5),
+                            ('c', 3);
+    `);
+
+    db.exec(`
+      CREATE TRIGGER test_trigger
+      AFTER INSERT ON t3
+      BEGIN
+          SELECT func(1) OVER ();
+      END;
+    `);
+
+    // TRIGGER will work fine with the window function
+    db.exec('INSERT INTO t3 VALUES(\'d\', 6)');
+  });
+
+  test('set SQLITE_DIRECT_ONLY flag when true', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.aggregate('func', {
+      start: 0,
+      step: (acc, value) => acc + value,
+      inverse: (acc, value) => acc - value,
+      directOnly: true,
+    });
+    db.exec(`
+      CREATE TABLE t3(x, y);
+      INSERT INTO t3 VALUES ('a', 4),
+                            ('b', 5),
+                            ('c', 3);
+    `);
+
+    db.exec(`
+      CREATE TRIGGER test_trigger
+      AFTER INSERT ON t3
+      BEGIN
+          SELECT func(1) OVER ();
+      END;
+    `);
+
+    t.assert.throws(() => {
+      db.exec('INSERT INTO t3 VALUES(\'d\', 6)');
+    }, {
+      code: 'ERR_SQLITE_ERROR',
+      message: /unsafe use of func\(\)/
+    });
+  });
+});
+
+describe('start', () => {
+  test('start option as a value', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.exec('CREATE TABLE data (value INTEGER)');
+    db.exec('INSERT INTO data VALUES (1), (2), (3)');
+    db.aggregate('sum_int', {
+      start: 0,
+      step: (acc, value) => acc + value,
+    });
+
+    const result = db.prepare('SELECT sum_int(value) as total FROM data').get();
+
+    t.assert.deepStrictEqual(result, { __proto__: null, total: 6 });
+  });
+
+  test('start option as a function', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.exec('CREATE TABLE data (value INTEGER)');
+    db.exec('INSERT INTO data VALUES (1), (2), (3)');
+    db.aggregate('sum_int', {
+      start: () => 0,
+      step: (acc, value) => acc + value,
+    });
+
+    const result = db.prepare('SELECT sum_int(value) as total FROM data').get();
+
+    t.assert.deepStrictEqual(result, { __proto__: null, total: 6 });
+  });
+
+  test('start option can hold any js value', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.exec('CREATE TABLE data (value INTEGER)');
+    db.exec('INSERT INTO data VALUES (1), (2), (3)');
+    db.aggregate('sum_int', {
+      start: () => [],
+      step: (acc, value) => {
+        return [...acc, value];
+      },
+      result: (acc) => acc.join(', '),
+    });
+
+    const result = db.prepare('SELECT sum_int(value) as total FROM data').get();
+
+    t.assert.deepStrictEqual(result, { __proto__: null, total: '1, 2, 3' });
+  });
+
+  test('throws if start throws an error', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.exec('CREATE TABLE data (value INTEGER)');
+    db.exec('INSERT INTO data VALUES (1), (2), (3)');
+    db.aggregate('agg', {
+      start: () => {
+        throw new Error('start error');
+      },
+      step: () => null,
+    });
+
+    t.assert.throws(() => {
+      db.prepare('SELECT agg()').get();
+    }, {
+      message: 'start error'
+    });
+  });
+});
+
+describe('step', () => {
+  test('throws if step throws an error', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    db.exec('CREATE TABLE data (value INTEGER)');
+    db.exec('INSERT INTO data VALUES (1), (2), (3)');
+    db.aggregate('agg', {
+      start: 0,
+      step: () => {
+        throw new Error('step error');
+      },
+    });
+
+    t.assert.throws(() => {
+      db.prepare('SELECT agg()').get();
+    }, {
+      message: 'step error'
+    });
+  });
+});
+
+describe('result', () => {
+  test('executes once when options.inverse is not present', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    const mockFn = t.mock.fn(() => 'overridden');
+    db.exec('CREATE TABLE data (value INTEGER)');
+    db.exec('INSERT INTO data VALUES (1), (2), (3)');
+    db.aggregate('sum_int', {
+      start: 0,
+      step: (acc, value) => {
+        return acc + value;
+      },
+      result: mockFn
+    });
+
+    const result = db.prepare('SELECT sum_int(value) as result FROM data').get();
+
+    t.assert.deepStrictEqual(result, { __proto__: null, result: 'overridden' });
+    t.assert.strictEqual(mockFn.mock.calls.length, 1);
+    t.assert.deepStrictEqual(mockFn.mock.calls[0].arguments, [6]);
+  });
+
+  test('executes once per row when options.inverse is present', (t) => {
+    const db = new DatabaseSync(':memory:');
+    t.after(() => db.close());
+    const mockFn = t.mock.fn((acc) => acc);
+    db.exec(`
+      CREATE TABLE t3(x, y);
+      INSERT INTO t3 VALUES ('a', 4),
+                            ('b', 5),
+                            ('c', 3);
+    `);
+    db.aggregate('sumint', {
+      start: 0,
+      step: (acc, value) => {
+        return acc + value;
+      },
+      inverse: (acc, value) => {
+        return acc - value;
+      },
+      result: mockFn
+    });
+
+    db.prepare(`
+      SELECT x, sumint(y) OVER (
+        ORDER BY x ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING
+      ) AS sum_y
+      FROM t3 ORDER BY x;
+    `).all();
+
+    t.assert.strictEqual(mockFn.mock.calls.length, 3);
+    t.assert.deepStrictEqual(mockFn.mock.calls[0].arguments, [9]);
+    t.assert.deepStrictEqual(mockFn.mock.calls[1].arguments, [12]);
+    t.assert.deepStrictEqual(mockFn.mock.calls[2].arguments, [8]);
+  });
+});
+
+test('throws an error when trying to use as windown function but didn\'t provide options.inverse', (t) => {
+  const db = new DatabaseSync(':memory:');
+  t.after(() => db.close());
+  db.exec(`
+    CREATE TABLE t3(x, y);
+    INSERT INTO t3 VALUES ('a', 4),
+                          ('b', 5),
+                          ('c', 3);
+  `);
+
+  db.aggregate('sumint', {
+    start: 0,
+    step: (total, nextValue) => total + nextValue,
+  });
+
+  t.assert.throws(() => {
+    db.prepare(`
+      SELECT x, sumint(y) OVER (
+        ORDER BY x ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING
+      ) AS sum_y
+      FROM t3 ORDER BY x;
+    `);
+  }, {
+    code: 'ERR_SQLITE_ERROR',
+    message: 'sumint() may not be used as a window function'
+  });
+});