executable.

apply executable @ select.

test suite prep.

??

best bench results.

...

update qb.

merge qb.

insert qb.

delete qb.

kysely.

raw builder.

fix regression in takeFirst.

more core.

abort.

...

...

...

...

...

...

...

...

...

executeQuery.

...

...

...

...

...

...

...

...

Author: igalklebanov

src/dialect/postgres/postgres-driver.ts

@@ -1,11 +1,14 @@
 import {
+  ControlConnectionProvider,
   DatabaseConnection,
+  QueryOptions,
   QueryResult,
 } from '../../driver/database-connection.js'
 import { Driver, TransactionSettings } from '../../driver/driver.js'
 import { parseSavepointCommand } from '../../parser/savepoint-parser.js'
 import { CompiledQuery } from '../../query-compiler/compiled-query.js'
 import { QueryCompiler } from '../../query-compiler/query-compiler.js'
+import { logOnce } from '../../util/log-once.js'
 import { isFunction, freeze } from '../../util/object-utils.js'
 import { createQueryId } from '../../util/query-id.js'
 import { extendStackTrace } from '../../util/stack-trace-utils.js'
@@ -144,20 +147,45 @@ interface PostgresConnectionOptions {
 }
 
 class PostgresConnection implements DatabaseConnection {
-  #client: PostgresPoolClient
-  #options: PostgresConnectionOptions
+  readonly #client: PostgresPoolClient
+  readonly #options: PostgresConnectionOptions
+  #pid: unknown
 
   constructor(client: PostgresPoolClient, options: PostgresConnectionOptions) {
     this.#client = client
     this.#options = options
   }
 
-  async executeQuery<O>(compiledQuery: CompiledQuery): Promise<QueryResult<O>> {
-    try {
-      const { command, rowCount, rows } = await this.#client.query<O>(
-        compiledQuery.sql,
-        [...compiledQuery.parameters],
+  async cancelQuery(
+    controlConnectionProvider: ControlConnectionProvider,
+  ): Promise<void> {
+    if (!this.#pid) {
+      return logOnce(
+        'kysely:warning: cannot cancel query because the connection has not been made cancelable.',
+      )
+    }
+
+    return await controlConnectionProvider(async (controlConnection) => {
+      await controlConnection.executeQuery(
+        CompiledQuery.raw('select pg_cancel_backend($1)', [this.#pid]),
       )
+    })
+  }
+
+  async executeQuery<O>(
+    compiledQuery: CompiledQuery,
+    options?: QueryOptions,
+  ): Promise<QueryResult<O>> {
+    try {
+      if (options?.cancelable) {
+        await this.#setupCancelability()
+      }
+
+      const result = await this.#client.query<O>(compiledQuery.sql, [
+        ...compiledQuery.parameters,
+      ])
+
+      const { command, rowCount, rows } = result
 
       return {
         numAffectedRows:
@@ -177,6 +205,7 @@ class PostgresConnection implements DatabaseConnection {
   async *streamQuery<O>(
     compiledQuery: CompiledQuery,
     chunkSize: number,
+    options?: QueryOptions,
   ): AsyncIterableIterator<QueryResult<O>> {
     if (!this.#options.cursor) {
       throw new Error(
@@ -188,6 +217,10 @@ class PostgresConnection implements DatabaseConnection {
       throw new Error('chunkSize must be a positive integer')
     }
 
+    if (options?.cancelable) {
+      await this.#setupCancelability()
+    }
+
     const cursor = this.#client.query(
       new this.#options.cursor<O>(
         compiledQuery.sql,
@@ -215,4 +248,19 @@ class PostgresConnection implements DatabaseConnection {
   [PRIVATE_RELEASE_METHOD](): void {
     this.#client.release()
   }
+
+  async #setupCancelability(): Promise<void> {
+    if (this.#pid) {
+      return
+    }
+
+    const {
+      rows: [row],
+    } = await this.#client.query<{ pid: unknown }>(
+      'select pg_backend_pid() as pid',
+      [],
+    )
+
+    this.#pid = row.pid
+  }
 }

src/driver/database-connection.ts

@@ -6,13 +6,30 @@ import { CompiledQuery } from '../query-compiler/compiled-query.js'
  * These are created by an instance of {@link Driver}.
  */
 export interface DatabaseConnection {
-  executeQuery<R>(compiledQuery: CompiledQuery): Promise<QueryResult<R>>
+  cancelQuery?(
+    controlConnectionProvider: ControlConnectionProvider,
+  ): Promise<void>
+
+  executeQuery<R>(
+    compiledQuery: CompiledQuery,
+    options?: QueryOptions,
+  ): Promise<QueryResult<R>>
+
   streamQuery<R>(
     compiledQuery: CompiledQuery,
-    chunkSize?: number,
+    chunkSize: number,
+    options?: QueryOptions,
   ): AsyncIterableIterator<QueryResult<R>>
 }
 
+export interface QueryOptions {
+  cancelable?: boolean
+}
+
+export type ControlConnectionProvider = (
+  consumer: (connection: DatabaseConnection) => Promise<void>,
+) => Promise<void>
+
 export interface QueryResult<O> {
   /**
    * This is defined for insert, update, delete and merge queries and contains

src/driver/runtime-driver.ts

@@ -167,12 +167,13 @@ export class RuntimeDriver implements Driver {
 
     connection.executeQuery = async (
       compiledQuery,
+      options,
     ): Promise<QueryResult<any>> => {
       let caughtError: unknown
       const startTime = performanceNow()
 
       try {
-        return await executeQuery.call(connection, compiledQuery)
+        return await executeQuery.call(connection, compiledQuery, options)
       } catch (error) {
         caughtError = error
         await dis.#logError(error, compiledQuery, startTime)
@@ -187,6 +188,7 @@ export class RuntimeDriver implements Driver {
     connection.streamQuery = async function* (
       compiledQuery,
       chunkSize,
+      options,
     ): AsyncIterableIterator<QueryResult<any>> {
       let caughtError: unknown
       const startTime = performanceNow()
@@ -196,6 +198,7 @@ export class RuntimeDriver implements Driver {
           connection,
           compiledQuery,
           chunkSize,
+          options,
         )) {
           yield result
         }

src/index.ts

@@ -224,6 +224,7 @@ export * from './util/column-type.js'
 export * from './util/compilable.js'
 export * from './util/explainable.js'
 export * from './util/streamable.js'
+export * from './util/executable.js'
 export * from './util/log.js'
 export {
   AnyAliasedColumn,
@@ -240,6 +241,7 @@ export {
 export * from './util/infer-result.js'
 export { logOnce } from './util/log-once.js'
 export { createQueryId, QueryId } from './util/query-id.js'
+export * from './util/abort.js'
 
 export {
   SelectExpression,

src/kysely.ts

@@ -2,7 +2,10 @@ import { Dialect } from './dialect/dialect.js'
 import { SchemaModule } from './schema/schema.js'
 import { DynamicModule } from './dynamic/dynamic.js'
 import { DefaultConnectionProvider } from './driver/default-connection-provider.js'
-import { QueryExecutor } from './query-executor/query-executor.js'
+import {
+  ExecuteQueryOptions,
+  QueryExecutor,
+} from './query-executor/query-executor.js'
 import { QueryCreator, QueryCreatorProps } from './query-creator.js'
 import { KyselyPlugin } from './plugin/kysely-plugin.js'
 import { DefaultQueryExecutor } from './query-executor/default-query-executor.js'
@@ -48,7 +51,7 @@ import {
   provideControlledConnection,
 } from './util/provide-controlled-connection.js'
 import { ConnectionProvider } from './driver/connection-provider.js'
-import { logOnce } from './util/log-once.js'
+import { ExecuteOptions } from './util/executable.js'
 
 // @ts-ignore
 Symbol.asyncDispose ??= Symbol('Symbol.asyncDispose')
@@ -538,20 +541,13 @@ export class Kysely<DB>
    *
    * See {@link https://github.com/kysely-org/kysely/blob/master/site/docs/recipes/0004-splitting-query-building-and-execution.md#execute-compiled-queries splitting build, compile and execute code recipe} for more information.
    */
-  executeQuery<R>(
+  async executeQuery<R>(
     query: CompiledQuery<R> | Compilable<R>,
-    // TODO: remove this in the future. deprecated in  0.28.x
-    queryId?: QueryId,
+    options?: ExecuteOptions,
   ): Promise<QueryResult<R>> {
-    if (queryId !== undefined) {
-      logOnce(
-        'Passing `queryId` in `db.executeQuery` is deprecated and will result in a compile-time error in the future.',
-      )
-    }
-
     const compiledQuery = isCompilable(query) ? query.compile() : query
 
-    return this.getExecutor().executeQuery<R>(compiledQuery)
+    return await this.getExecutor().executeQuery<R>(compiledQuery, options)
   }
 
   async [Symbol.asyncDispose]() {
@@ -1177,17 +1173,21 @@ class NotCommittedOrRolledBackAssertingExecutor implements QueryExecutor {
     return this.#executor.provideConnection(consumer)
   }
 
-  executeQuery<R>(compiledQuery: CompiledQuery<R>): Promise<QueryResult<R>> {
+  executeQuery<R>(
+    compiledQuery: CompiledQuery<R>,
+    options?: ExecuteQueryOptions,
+  ): Promise<QueryResult<R>> {
     assertNotCommittedOrRolledBack(this.#state)
-    return this.#executor.executeQuery(compiledQuery)
+    return this.#executor.executeQuery(compiledQuery, options)
   }
 
   stream<R>(
     compiledQuery: CompiledQuery<R>,
     chunkSize: number,
+    options?: ExecuteQueryOptions,
   ): AsyncIterableIterator<QueryResult<R>> {
     assertNotCommittedOrRolledBack(this.#state)
-    return this.#executor.stream(compiledQuery, chunkSize)
+    return this.#executor.stream(compiledQuery, chunkSize, options)
   }
 
   withConnectionProvider(