diff --git a/README.md b/README.md index f8c649f..c51fdff 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,31 @@ importModel(DbBackend.postgre, "model_postgre") let db {.global.} = open("localhost", "user", "password", "dbname") ``` +For repeated queries on a dedicated PostgreSQL session, use the opt-in owned +connection. It prepares each distinct SQL statement once on that session and +reuses it. The cache defaults to 64 plans; additional distinct queries execute +without creating another server-side plan. Set `maxPlans = 0` to disable the +cache. The raw `DbConn` API remains available. + +```nim +var db = openManaged("host=localhost dbname=dbname user=user password=password") +defer: db.close() # optional for a local variable: scope exit closes it + +let rows = query: + select tb_serial(typinteger) +``` + +`ManagedDbConn` is move-only, so pass it as `var ManagedDbConn` to helpers. +`createProc` and `createIter` accept both raw and owned PostgreSQL connections. +Use one owned connection from one thread at a time. Named plans require a direct +connection or session pooler; transaction pooling can switch backends between +queries. `borrowRaw` provides a temporary handle for other libpq APIs; never +close or reset that handle. Use `db.reset()` to reconnect and clear the cache. +If a cached plan disappears or becomes invalid after a schema change, the query +raises its original database error and the cache is disabled for that session. +Ormin does not replay a query after a connection or plan error because its +side effects may be uncertain. Callers may reset and retry when appropriate. + ## Query DSL `query:` blocks are turned into prepared statements at compile time. Placeholders use `?` for Nim values and `%` for JSON values; Ormin chooses JSON instead of an ad-hoc variant type so your data can flow straight from/into `JsonNode` trees. `!!` splices vendor-specific SQL fragments. Typical clauses such as `with`, `where`, joins, `orderby`, `groupby`, `limit`, `offset`, `exists`, `distinct`, window expressions, `union`/`intersect`/`except`, `returning`, and insert upserts via `onconflict` + (`donothing` or `doupdate`) are supported. Referring to columns from related tables can trigger **automatic join generation** based on foreign keys, reducing boilerplate joins. diff --git a/config.nims b/config.nims index 83704b0..a575e88 100644 --- a/config.nims +++ b/config.nims @@ -35,9 +35,12 @@ task test_postgres, "Run PostgreSQL test suite": # These suites share one PostgreSQL database and must execute serially. exec("atlas-run tests --jobs:1 tests/tfeature.nim tests/tcommon.nim " & - "tests/tpostgre.nim -- -d:nimDebugDlOpen -d:postgre") + "tests/tpostgre.nim tests/tpostgre_statement_lifecycle.nim " & + "tests/tpostgre_managed_cache.nim -- " & + "-d:nimDebugDlOpen -d:postgre") exec("atlas-run tests --jobs:2 tests/tquery_types.nim -- " & "-d:nimDebugDlOpen -d:release -d:postgre") + exec "bash tests/check_managed_copy.sh" task buildexamples, "Build examples: chat and forum": buildimporterTask() diff --git a/ormin/ormin_postgre.nim b/ormin/ormin_postgre.nim index e3f1beb..e02e21c 100644 --- a/ormin/ormin_postgre.nim +++ b/ormin/ormin_postgre.nim @@ -6,7 +6,18 @@ export db_common type DbConn* = PPGconn ## encapsulates a database connection - PStmt = string ## a identifier for the prepared queries + PStmt = string ## SQL text passed to the extended query protocol + + ManagedPlan = object + sql, name: string + + ManagedDbConn* = object + ## Owns one libpq session and at most `maxPlans` named plans in that session. + raw: DbConn + plans: seq[ManagedPlan] + maxPlans, nextPlanId: int + backendPid: int32 + cacheDisabled: bool varcharType* = string intType* = int @@ -16,6 +27,20 @@ type serialType* = int jsonType* = JsonNode +proc `=destroy`(db: ManagedDbConn) = + if db.raw != nil: + pqfinish(db.raw) + `=destroy`(db.plans) + +proc `=wasMoved`(db: var ManagedDbConn) = + db.raw = nil + wasMoved(db.plans) + +proc `=copy`(dest: var ManagedDbConn; source: ManagedDbConn) {.error.} +proc `=dup`(source: ManagedDbConn): ManagedDbConn {.error.} + +const DEFAULT_MANAGED_PLAN_LIMIT* = 64 + var jsonTimeFormat* = "yyyy-MM-dd HH:mm:ss" proc dbError*(db: DbConn) {.noreturn.} = @@ -25,21 +50,26 @@ proc dbError*(db: DbConn) {.noreturn.} = e.msg = $pqErrorMessage(db) raise e +proc dbError*(db: var ManagedDbConn) {.noreturn.} = + ## Raises the same PostgreSQL error as the raw connection, or a clear + ## error when the owned connection has already been closed. + if db.raw == nil: + raise newException(DbError, "Managed PostgreSQL connection is closed") + dbError(db.raw) + proc c_strtod(buf: cstring, endptr: ptr cstring = nil): float64 {. importc: "strtod", header: "", noSideEffect.} proc c_strtol(buf: cstring, endptr: ptr cstring = nil, base: cint = 10): int {. importc: "strtol", header: "", noSideEffect.} -var sid {.compileTime.}: int - proc prepareStmt*(db: DbConn; q: string): PStmt = when defined(debugOrminTrace): echo "[[Ormin Executing]]: ", q - inc sid - result = "ormin" & $sid - var res = pqprepare(db, result, q, 0, nil) - if pqResultStatus(res) != PGRES_COMMAND_OK: dbError(db) + # Ormin used to create a new named server-side statement for every call. + # Long-lived connections retained every plan, and the prepare result was + # never cleared. Execute parameterized SQL directly instead. + result = q template startBindings*(s: PStmt; n: int) {.dirty.} = # pparams is a duplicated array to keep the Nim string alive @@ -232,16 +262,27 @@ template bindToJson*(db: DbConn; s: PStmt; idx: int; obj: JsonNode; let src = pqgetvalue(queryResult, queryI, idx.cint) obj[name] = parseJson($src) +proc checkQueryResult(db: DbConn; res: PPGresult) = + if res != nil and pqResultStatus(res) in {PGRES_COMMAND_OK, PGRES_TUPLES_OK}: + return + # Copy the result's error before releasing it. Failed PQexecParams calls + # still return an allocated PGresult that the caller will never stopQuery. + let detail = + if res == nil: $pqErrorMessage(db) + else: $pqResultErrorMessage(res) + let message = if detail.len > 0: detail else: $pqErrorMessage(db) + if res != nil: + pqclear(res) + raise newException(DbError, message) + template startQuery*(db: DbConn; s: PStmt) = when declared(pparams): - var queryResult {.inject.} = pqexecPrepared(db, s, int32(parr.len), - cast[cstringArray](addr parr), nil, nil, 0) + var queryResult {.inject.} = pqexecParams(db, s.cstring, int32(parr.len), + nil, cast[cstringArray](addr parr), nil, nil, 0) else: - var queryResult {.inject.} = pqexecPrepared(db, s, int32(0), - nil, nil, nil, 0) - if pqResultStatus(queryResult) == PGRES_COMMAND_OK: - discard # insert does not returns data in pg - elif pqResultStatus(queryResult) != PGRES_TUPLES_OK: ormin_postgre.dbError(db) + var queryResult {.inject.} = pqexecParams(db, s.cstring, int32(0), + nil, nil, nil, nil, 0) + checkQueryResult(db, queryResult) var queryI {.inject.} = cint(-1) var queryLen {.inject.} = pqntuples(queryResult) @@ -272,3 +313,162 @@ proc open*(connection, user, password, database: string): DbConn = else: substr(connection, colonPos+1) result = pqsetdbLogin(host, port, nil, nil, database, user, password) if pqStatus(result) != CONNECTION_OK: dbError(result) # result = nil + +proc openManaged*(conninfo: string; + maxPlans = DEFAULT_MANAGED_PLAN_LIMIT): ManagedDbConn = + ## Opens an owned session. Use a direct PostgreSQL or session-pooled connection; + ## transaction pooling cannot promise that named plans stay on one backend. + if maxPlans < 0: + raise newException(ValueError, "maxPlans must be nonnegative") + result.raw = pqconnectdb(conninfo.cstring) + if result.raw == nil: + raise newException(DbError, "Could not allocate a PostgreSQL connection") + if pqStatus(result.raw) != CONNECTION_OK: + let message = $pqErrorMessage(result.raw) + pqfinish(result.raw) + result.raw = nil + raise newException(DbError, message) + result.maxPlans = maxPlans + result.backendPid = pqbackendPID(result.raw) + +proc borrowRaw*(db: var ManagedDbConn): DbConn = + ## Borrows the libpq handle for APIs that require it. Do not finish or reset + ## this handle; use `close` or `reset` on the owner instead. + db.raw + +proc managedPlanCount*(db: var ManagedDbConn): int = db.plans.len + +proc close*(db: var ManagedDbConn) = + ## Closes the session and discards its cache. Repeated close is harmless. + `=destroy`(db) + `=wasMoved`(db) + +proc reset*(db: var ManagedDbConn) = + ## Reconnects the same libpq handle and invalidates every cached plan. + if db.raw == nil: + raise newException(DbError, "Managed PostgreSQL connection is closed") + db.plans.setLen(0) + db.nextPlanId = 0 + db.cacheDisabled = false + pqreset(db.raw) + if pqStatus(db.raw) != CONNECTION_OK: + dbError(db.raw) + db.backendPid = pqbackendPID(db.raw) + +proc planName(db: var ManagedDbConn; sql: string): string = + for plan in db.plans: + if plan.sql == sql: + return plan.name + +proc prepareStmt*(db: var ManagedDbConn; q: string): PStmt = + ## Prepares a query once per owned session, up to its plan limit. Excess + ## distinct queries use PQexecParams without retaining another server plan. + if db.raw == nil: + raise newException(DbError, "Managed PostgreSQL connection is closed") + if pqStatus(db.raw) != CONNECTION_OK: + dbError(db.raw) + let pid = pqbackendPID(db.raw) + if pid != db.backendPid: + db.plans.setLen(0) + db.nextPlanId = 0 + db.backendPid = pid + when defined(debugOrminTrace): + echo "[[Ormin Executing]]: ", q + result = q + if db.cacheDisabled or db.maxPlans == 0: + return + if db.planName(q).len > 0: + return + if db.plans.len >= db.maxPlans: + return + let name = "ormin_managed_" & $db.nextPlanId + let res = pqprepare(db.raw, name.cstring, q.cstring, 0, nil) + if res == nil: + dbError(db.raw) + let status = pqResultStatus(res) + if status != PGRES_COMMAND_OK: + let detail = $pqResultErrorMessage(res) + let message = if detail.len > 0: detail else: $pqErrorMessage(db.raw) + pqclear(res) + raise newException(DbError, message) + pqclear(res) + inc db.nextPlanId + db.plans.add ManagedPlan(sql: q, name: name) + +proc executeManaged(db: var ManagedDbConn; sql: string; + params: cstringArray; count: int32): PPGresult = + if db.raw == nil: + raise newException(DbError, "Managed PostgreSQL connection is closed") + let name = db.planName(sql) + if name.len > 0 and not db.cacheDisabled: + result = pqexecPrepared(db.raw, name.cstring, count, params, nil, nil, 0) + else: + result = pqexecParams(db.raw, sql.cstring, count, nil, params, nil, nil, 0) + if result == nil: + dbError(db.raw) + if pqResultStatus(result) in {PGRES_COMMAND_OK, PGRES_TUPLES_OK}: + return + if name.len > 0 and not db.cacheDisabled: + let sqlState = $pqResultErrorField(result, int32(ord('C'))) + if sqlState in ["26000", "0A000"]: + # The server lost this plan or changed its result shape. Never replay an + # uncertain operation, and use direct execution for this session now. + db.cacheDisabled = true + db.plans.setLen(0) + let detail = $pqResultErrorMessage(result) + let message = if detail.len > 0: detail else: $pqErrorMessage(db.raw) + pqclear(result) + result = nil + raise newException(DbError, message) + +template startQuery*(db: ManagedDbConn; s: PStmt) = + when declared(pparams): + var queryResult {.inject.} = executeManaged(db, s, + cast[cstringArray](addr parr), int32(parr.len)) + else: + var queryResult {.inject.} = executeManaged(db, s, nil, 0) + # Generated row conversion can raise after a successful execution. Release + # that result even when the normal stopQuery path is skipped. + defer: + if queryResult != nil: + pqclear(queryResult) + var queryI {.inject.} = cint(-1) + var queryLen {.inject.} = pqntuples(queryResult) + +template bindParam*(db: ManagedDbConn; s: PStmt; idx: int; x: untyped; t: untyped) = + bindParam(db.raw, s, idx, x, t) + +template bindNullParam*(db: ManagedDbConn; s: PStmt; idx: int) = + bindNullParam(db.raw, s, idx) + +template bindParamJson*(db: ManagedDbConn; s: PStmt; idx: int; + x: JsonNode; t: typedesc) = + bindParamJson(db.raw, s, idx, x, t) + +template bindResult*(db: ManagedDbConn; s: PStmt; idx: int; + dest: untyped; t: typedesc; name: string) = + bindResult(db.raw, s, idx, dest, t, name) + +template bindResultJson*(db: ManagedDbConn; s: PStmt; idx: int; + obj: JsonNode; t: typedesc; name: string) = + bindResultJson(db.raw, s, idx, obj, t, name) + +template columnIsNull*(db: ManagedDbConn; s: PStmt; idx: int): bool = + columnIsNull(db.raw, s, idx) + +template bindToJson*(db: ManagedDbConn; s: PStmt; idx: int; + obj: JsonNode; t: typedesc; name: string) = + bindToJson(db.raw, s, idx, obj, t, name) + +template stopQuery*(db: ManagedDbConn; s: PStmt) = + pqclear(queryResult) + queryResult = nil + +template stepQuery*(db: ManagedDbConn; s: PStmt; returnsData: bool): bool = + stepQuery(db.raw, s, returnsData) + +template getLastId*(db: ManagedDbConn; s: PStmt): int = + getLastId(db.raw, s) + +template getAffectedRows*(db: ManagedDbConn; s: PStmt): int = + getAffectedRows(db.raw, s) diff --git a/ormin/queries.nim b/ormin/queries.nim index f2426b9..d0c5962 100644 --- a/ormin/queries.nim +++ b/ormin/queries.nim @@ -822,7 +822,7 @@ proc addStatementCleanup(body, statement: NimNode) = defer: finalizeStmt(`statement`) proc generateRoutine(name: NimNode, q: QueryBuilder; - sql: string; k: NimNodeKind): NimNode = + sql: string; k: NimNodeKind; dbType: NimNode): NimNode = let prepStmt = ident($name & "PrepStmt") let prepare = newVarStmt(prepStmt, newCall(bindSym"prepareStmt", ident"db", newLit(sql))) @@ -847,12 +847,7 @@ proc generateRoutine(name: NimNode, q: QueryBuilder; if k != nnkIteratorDef: rtyp = nnkBracketExpr.newTree(ident"seq", rtyp) finalParams.add rtyp - when dbBackend == DbBackend.postgre: - finalParams.add newIdentDefs(ident"db", newTree(nnkDotExpr, ident"ormin_postgre", ident"DbConn")) - elif dbBackend == DbBackend.sqlite: - finalParams.add newIdentDefs(ident"db", newTree(nnkDotExpr, ident"ormin_sqlite", ident"DbConn")) - else: - finalParams.add newIdentDefs(ident"db", ident("DbConn")) + finalParams.add newIdentDefs(ident"db", dbType) var i = 1 if q.params.len > 0: body.add newCall(bindSym"startBindings", prepStmt, newLit(q.params.len)) @@ -1938,7 +1933,17 @@ proc createRoutine(name, query: NimNode; k: NimNodeKind): NimNode = if q.kind notin {qkSelect, qkJoin}: macros.error "query must be a 'select' or 'join'", query let sql = queryAsString(q, query) - result = generateRoutine(name, q, sql, k) + when dbBackend == DbBackend.postgre: + let raw = newTree(nnkDotExpr, ident"ormin_postgre", ident"DbConn") + let managed = newTree(nnkVarTy, + newTree(nnkDotExpr, ident"ormin_postgre", ident"ManagedDbConn")) + result = newStmtList(generateRoutine(name, q, sql, k, raw), + generateRoutine(name, q, sql, k, managed)) + elif dbBackend == DbBackend.sqlite: + let dbType = newTree(nnkDotExpr, ident"ormin_sqlite", ident"DbConn") + result = generateRoutine(name, q, sql, k, dbType) + else: + result = generateRoutine(name, q, sql, k, ident("DbConn")) when defined(debugOrminDsl): macros.hint("Ormin Query: " & repr(result), query) diff --git a/tests/check_managed_copy.sh b/tests/check_managed_copy.sh new file mode 100644 index 0000000..687f555 --- /dev/null +++ b/tests/check_managed_copy.sh @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +set -euo pipefail + +output=$(nim c tests/should_not_compile_managed_copy.nim 2>&1) && { + echo "ManagedDbConn unexpectedly allowed a copy" + exit 1 +} + +if [[ "$output" != *"'=copy' is not available for type "* ]]; then + echo "$output" + exit 1 +fi diff --git a/tests/config.nims b/tests/config.nims index 1bc058e..aa3c1b9 100644 --- a/tests/config.nims +++ b/tests/config.nims @@ -1 +1 @@ ---path:"$projectDir/../../ormin" \ No newline at end of file +--path:"$projectDir/.." diff --git a/tests/should_not_compile_managed_copy.nim b/tests/should_not_compile_managed_copy.nim new file mode 100644 index 0000000..73d343e --- /dev/null +++ b/tests/should_not_compile_managed_copy.nim @@ -0,0 +1,9 @@ +import ormin/ormin_postgre + +proc copyAnOwnedConnection() = + var original: ManagedDbConn + var copied = original + discard original.borrowRaw + discard copied.borrowRaw + +copyAnOwnedConnection() diff --git a/tests/tpostgre_managed_cache.nim b/tests/tpostgre_managed_cache.nim new file mode 100644 index 0000000..510193c --- /dev/null +++ b/tests/tpostgre_managed_cache.nim @@ -0,0 +1,356 @@ +import std/[os, strutils, unittest] + +import db_connector/db_postgres as pg +from db_connector/postgres import PPGconn, pqgetvalue, pqnfields +import ormin +import ormin/ormin_postgre as orminPostgre + +importModel(DbBackend.postgre, "model_postgre") + +let target = getEnv("TEST_DATABASE_URL", "postgresql://test@localhost/test_ormin") + +when defined(windows): + const libpq = "libpq.dll" +elif defined(macosx): + const libpq = "libpq.dylib" +else: + const libpq = "libpq.so(.5|)" + +type ResultEventProc = proc(event: cint; info, context: pointer): cint {.cdecl.} + +proc registerResultEvent(conn: PPGconn; callback: ResultEventProc; + name: cstring; context: pointer): cint {. + cdecl, importc: "PQregisterEventProc", dynlib: libpq.} + +var resultCreated, resultDestroyed: int + +proc countResultEvent(event: cint; info, context: pointer): cint {.cdecl.} = + # libpq reports PGresult creation/destruction as event IDs 3 and 5. + if event == 3: + inc resultCreated + elif event == 5: + inc resultDestroyed + 1 + +type FailingConversion = distinct int + +proc fromQueryHook*(dest: var FailingConversion; value: int) = + raise newException(ValueError, "conversion failed") + +proc queryInt(db: var ManagedDbConn; sql: string; value: int): string = + let statement = orminPostgre.prepareStmt(db, sql) + orminPostgre.startBindings(statement, 1) + orminPostgre.bindParam(db, statement, 1, value, int) + orminPostgre.startQuery(db, statement) + doAssert orminPostgre.stepQuery(db, statement, true) + result = $pqgetvalue(queryResult, queryI, 0) + orminPostgre.stopQuery(db, statement) + +proc querySerial(db: var ManagedDbConn; value: int): seq[int] = + result = query: + select tb_serial(typinteger) + where typinteger == ?value + +proc failAfterStart(db: var ManagedDbConn) = + let statement = orminPostgre.prepareStmt(db, "select 1") + orminPostgre.startQuery(db, statement) + raise newException(ValueError, "conversion failed") + +proc scopedBackendPid(target: string): int = + var db = openManaged(target) + discard queryInt(db, "select $1::integer", 7) + result = pg.getValue(db.borrowRaw, sql"select pg_backend_pid()").parseInt() + +suite "owned PostgreSQL plan cache": + test "same SQL uses one plan, while excess queries execute directly": + var db = openManaged(target, maxPlans = 1) + for value in 0 ..< 100: + check queryInt(db, "select $1::integer", value) == $value + check queryInt(db, "select $1::integer + 1", value) == $(value + 1) + check db.managedPlanCount == 1 + check pg.getValue(db.borrowRaw, sql"select count(*) from pg_prepared_statements") == "1" + db.close() + + test "zero plan limit uses direct execution": + var db = openManaged(target, maxPlans = 0) + for value in 0 ..< 25: + check queryInt(db, "select $1::integer", value) == $value + check db.managedPlanCount == 0 + check pg.getValue(db.borrowRaw, sql"select count(*) from pg_prepared_statements") == "0" + + test "move and explicit close transfer and release ownership": + var first = openManaged(target) + discard queryInt(first, "select $1::integer", 7) + pg.exec(first.borrowRaw, + sql"create temporary table tb_serial (typserial serial primary key, typinteger integer)") + pg.exec(first.borrowRaw, sql"insert into tb_serial(typinteger) values (17)") + var second = move(first) + check first.borrowRaw == nil + check queryInt(second, "select $1::integer", 8) == "8" + check querySerial(second, 17) == @[17] + second.close() + check second.borrowRaw == nil + second.close() + + test "move assignment releases the replaced session": + var first = openManaged(target) + var second = openManaged(target) + let replacedPid = pg.getValue(second.borrowRaw, sql"select pg_backend_pid()").parseInt() + second = move(first) + check first.borrowRaw == nil + check queryInt(second, "select $1::integer", 11) == "11" + let observer = pg.open("", "", "", target) + check pg.getValue(observer, sql("select count(*) from pg_stat_activity where pid = " & + $replacedPid)) == "0" + pg.close(observer) + second.close() + + test "closed connections reject queries and reset": + var db = openManaged(target) + db.close() + expect DbError: + discard orminPostgre.prepareStmt(db, "select 1") + expect DbError: + orminPostgre.dbError(db) + expect DbError: + db.reset() + + test "single-row queries and insert returning work on owned connections": + var db = openManaged(target) + pg.exec(db.borrowRaw, + sql"create temporary table tb_serial (typserial serial primary key, typinteger integer not null)") + let value = 31 + let insertedId = query: + insert tb_serial(typinteger = ?value) + returning typserial + check insertedId == 1 + + let untypedRow = query: + select tb_serial(typinteger) + where typserial == ?insertedId + limit 1 + check untypedRow == value + + let typedRow = query(int): + select tb_serial(typinteger) + where typserial == ?insertedId + limit 1 + check typedRow == value + + let missingId = 999 + expect DbError: + discard query: + select tb_serial(typinteger) + where typserial == ?missingId + limit 1 + expect DbError: + discard query(int): + select tb_serial(typinteger) + where typserial == ?missingId + limit 1 + let missingUntyped = tryQuery: + select tb_serial(typinteger) + where typserial == ?missingId + limit 1 + check missingUntyped == 0 + let missingTyped = tryQuery(int): + select tb_serial(typinteger) + where typserial == ?missingId + limit 1 + check missingTyped == 0 + + # A server error from INSERT ... RETURNING must keep the session usable. + expect DbError: + discard query: + insert tb_serial(typserial = ?insertedId, typinteger = ?value) + returning typserial + check pg.getValue(db.borrowRaw, sql"select count(*) from tb_serial") == "1" + check db.managedPlanCount <= DEFAULT_MANAGED_PLAN_LIMIT + + test "single-row success, empty results and errors release libpq results": + var db = openManaged(target) + pg.exec(db.borrowRaw, + sql"create temporary table tb_serial (typserial serial primary key, typinteger integer not null)") + check registerResultEvent(db.borrowRaw, countResultEvent, + "ormin-managed-result-lifetime", nil) == 1 + resultCreated = 0 + resultDestroyed = 0 + + let value = 7 + let insertedId = query: + insert tb_serial(typinteger = ?value) + returning typserial + check insertedId == 1 + let typedRow = query(int): + select tb_serial(typinteger) + where typserial == ?insertedId + limit 1 + check typedRow == value + let missingId = 999 + expect DbError: + discard query(int): + select tb_serial(typinteger) + where typserial == ?missingId + limit 1 + expect DbError: + discard query: + insert tb_serial(typserial = ?insertedId, typinteger = ?value) + returning typserial + expect ValueError: + discard query(FailingConversion): + select tb_serial(typinteger) + limit 1 + createIter rowsForResultCleanup: + select tb_serial(typinteger) + for row in db.rowsForResultCleanup(): + break + check resultCreated > 0 + check resultDestroyed == resultCreated + + test "generated query blocks use the owned connection cache": + var db = openManaged(target) + pg.exec(db.borrowRaw, + sql"create temporary table tb_serial (typserial serial primary key, typinteger integer not null)") + let value = 31 + query: + insert tb_serial(typinteger = ?value) + let rows = query: + select tb_serial(typinteger) + where typinteger == ?value + check rows == @[value] + for _ in 0 ..< 20: + let repeated = query: + select tb_serial(typinteger) + where typinteger == ?value + check repeated == @[value] + check db.managedPlanCount == 2 + check pg.getValue(db.borrowRaw, sql"select count(*) from pg_prepared_statements") == "2" + + createProc lookupSerial: + select tb_serial(typinteger) + where typinteger == ?value + createIter lookupSerialIter: + select tb_serial(typinteger) + where typinteger == ?value + for _ in 0 ..< 20: + check db.lookupSerial(value) == @[value] + var iterRows: seq[int] + for row in db.lookupSerialIter(value): + iterRows.add row + check iterRows == @[value] + check db.managedPlanCount == 2 + db.close() + + test "scope destruction closes the owned session": + let pid = scopedBackendPid(target) + let observer = pg.open("", "", "", target) + check pg.getValue(observer, sql("select count(*) from pg_stat_activity where pid = " & + $pid)) == "0" + pg.close(observer) + + test "reset discards plans and prepares them again on the new session": + var db = openManaged(target) + check queryInt(db, "select $1::integer", 2) == "2" + check db.managedPlanCount == 1 + db.reset() + check db.managedPlanCount == 0 + check pg.getValue(db.borrowRaw, sql"select count(*) from pg_prepared_statements") == "0" + check queryInt(db, "select $1::integer", 3) == "3" + check db.managedPlanCount == 1 + + test "generated queries prepare again after reset": + var db = openManaged(target) + pg.exec(db.borrowRaw, + sql"create temporary table tb_serial (typserial serial primary key, typinteger integer)") + pg.exec(db.borrowRaw, sql"insert into tb_serial(typinteger) values (18)") + check querySerial(db, 18) == @[18] + check db.managedPlanCount == 1 + db.reset() + check db.managedPlanCount == 0 + pg.exec(db.borrowRaw, + sql"create temporary table tb_serial (typserial serial primary key, typinteger integer)") + pg.exec(db.borrowRaw, sql"insert into tb_serial(typinteger) values (19)") + check querySerial(db, 19) == @[19] + check db.managedPlanCount == 1 + + test "reset recovers after a server-terminated connection": + var db = openManaged(target) + check queryInt(db, "select $1::integer", 2) == "2" + let pid = pg.getValue(db.borrowRaw, sql"select pg_backend_pid()").parseInt() + let observer = pg.open("", "", "", target) + check pg.getValue(observer, sql("select pg_terminate_backend(" & $pid & ")")) == "t" + pg.close(observer) + db.reset() + check db.managedPlanCount == 0 + check queryInt(db, "select $1::integer", 3) == "3" + + test "a failed bound value retains the plan and connection": + var db = openManaged(target) + let statement = orminPostgre.prepareStmt(db, "select $1::integer") + orminPostgre.startBindings(statement, 1) + orminPostgre.bindParam(db, statement, 1, "invalid", string) + var errorMessage = "" + try: + orminPostgre.startQuery(db, statement) + orminPostgre.stopQuery(db, statement) + except DbError as exc: + errorMessage = exc.msg + check "invalid input syntax" in errorMessage + check db.managedPlanCount == 1 + check queryInt(db, "select $1::integer", 4) == "4" + + test "a failed prepare preserves the server error and leaves the connection reusable": + var db = openManaged(target) + for _ in 0 ..< 25: + var errorMessage = "" + try: + discard orminPostgre.prepareStmt(db, "select from") + except DbError as exc: + errorMessage = exc.msg + check "syntax error" in errorMessage + check db.managedPlanCount == 0 + check queryInt(db, "select $1::integer", 9) == "9" + + test "unwinding after execution releases the result": + var db = openManaged(target) + for _ in 0 ..< 100: + expect ValueError: + failAfterStart(db) + check queryInt(db, "select $1::integer", 10) == "10" + + test "missing server plan disables the cache without replaying the query": + var db = openManaged(target) + check queryInt(db, "select $1::integer", 4) == "4" + pg.exec(db.borrowRaw, sql"deallocate all") + var errorMessage = "" + try: + discard queryInt(db, "select $1::integer", 5) + except DbError as exc: + errorMessage = exc.msg + check "does not exist" in errorMessage + check db.managedPlanCount == 0 + check queryInt(db, "select $1::integer", 6) == "6" + check pg.getValue(db.borrowRaw, sql"select count(*) from pg_prepared_statements") == "0" + + test "a schema change invalidates plans without replaying the query": + var db = openManaged(target) + pg.exec(db.borrowRaw, sql"create temporary table ormin_schema_test (a integer)") + let statement = orminPostgre.prepareStmt(db, "select * from ormin_schema_test") + block: + orminPostgre.startQuery(db, statement) + check pqnfields(queryResult) == 1 + orminPostgre.stopQuery(db, statement) + pg.exec(db.borrowRaw, sql"alter table ormin_schema_test add column b integer") + var errorMessage = "" + try: + block: + orminPostgre.startQuery(db, statement) + orminPostgre.stopQuery(db, statement) + except DbError as exc: + errorMessage = exc.msg + check "cached plan must not change result type" in errorMessage + check db.managedPlanCount == 0 + block: + orminPostgre.startQuery(db, statement) + check pqnfields(queryResult) == 2 + orminPostgre.stopQuery(db, statement) diff --git a/tests/tpostgre_statement_lifecycle.nim b/tests/tpostgre_statement_lifecycle.nim new file mode 100644 index 0000000..292141a --- /dev/null +++ b/tests/tpostgre_statement_lifecycle.nim @@ -0,0 +1,75 @@ +import std/[os, strutils, unittest] + +import db_connector/db_postgres as pg +from db_connector/postgres import pqgetvalue +import ormin/ormin_postgre as orminPostgre + +let target = getEnv("TEST_DATABASE_URL") +let db = + if target.len > 0: + pg.open("", "", "", target) + else: + pg.open("localhost", "test", "test", "test_ormin") + +suite "PostgreSQL statement lifecycle": + test "repeated parameterized queries leave no named plans behind": + let before = pg.getValue(db, sql"select count(*) from pg_prepared_statements").parseInt() + for value in 0 ..< 1000: + let statement = orminPostgre.prepareStmt(db, "select $1::integer") + orminPostgre.startBindings(statement, 1) + orminPostgre.bindParam(db, statement, 1, value, int) + orminPostgre.startQuery(db, statement) + check orminPostgre.stepQuery(db, statement, true) + check $pqgetvalue(queryResult, queryI, 0) == $value + orminPostgre.stopQuery(db, statement) + let after = pg.getValue(db, sql"select count(*) from pg_prepared_statements").parseInt() + check after == before + + test "queries without parameters and commands do not retain named plans": + let before = pg.getValue(db, sql"select count(*) from pg_prepared_statements").parseInt() + let selectStatement = orminPostgre.prepareStmt(db, "select 42") + orminPostgre.startQuery(db, selectStatement) + check orminPostgre.stepQuery(db, selectStatement, true) + check $pqgetvalue(queryResult, queryI, 0) == "42" + orminPostgre.stopQuery(db, selectStatement) + + pg.exec(db, sql"create temporary table ormin_lifecycle_test (value integer)") + for value in 0 ..< 100: + let insertStatement = orminPostgre.prepareStmt( + db, "insert into ormin_lifecycle_test (value) values ($1)") + orminPostgre.startBindings(insertStatement, 1) + orminPostgre.bindParam(db, insertStatement, 1, value, int) + orminPostgre.startQuery(db, insertStatement) + check orminPostgre.getAffectedRows(db, insertStatement) == 1 + orminPostgre.stopQuery(db, insertStatement) + check pg.getValue(db, sql"select count(*) from ormin_lifecycle_test") == "100" + pg.exec(db, sql"drop table ormin_lifecycle_test") + let after = pg.getValue(db, sql"select count(*) from pg_prepared_statements").parseInt() + check after == before + + test "a failed parameterized query keeps its error and leaves the connection usable": + let statement = orminPostgre.prepareStmt(db, "select $1::integer") + orminPostgre.startBindings(statement, 1) + orminPostgre.bindParam(db, statement, 1, "not an integer", string) + for attempt in 0 ..< 100: + var errorMessage = "" + try: + orminPostgre.startQuery(db, statement) + orminPostgre.stopQuery(db, statement) + except DbError as exc: + errorMessage = exc.msg + check "invalid input syntax" in errorMessage + check pg.getValue(db, sql"select 1") == "1" + + test "invalid SQL reports its error and does not poison the connection": + let statement = orminPostgre.prepareStmt(db, "select from") + var errorMessage = "" + try: + orminPostgre.startQuery(db, statement) + orminPostgre.stopQuery(db, statement) + except DbError as exc: + errorMessage = exc.msg + check "syntax error" in errorMessage + check pg.getValue(db, sql"select 1") == "1" + +pg.close(db)