diff --git a/composer.json b/composer.json
index 030922c..d6a4c33 100644
--- a/composer.json
+++ b/composer.json
@@ -22,18 +22,18 @@
         "php": ">=8.1",
         "amphp/amp": "^3",
         "amphp/pipeline": "^1",
-        "amphp/sql": "^2",
-        "amphp/sql-common": "^2"
+        "amphp/sql": "dev-refactor-transactions as 2.0",
+        "amphp/sql-common": "dev-refactor-transactions as 2.0"
     },
     "require-dev": {
         "ext-pgsql": "*",
         "ext-pq": "*",
         "amphp/phpunit-util": "^3",
         "phpunit/phpunit": "^9",
-        "amphp/php-cs-fixer-config": "^2-dev",
-        "psalm/phar": "^5.4"
+        "amphp/php-cs-fixer-config": "^2",
+        "psalm/phar": "^5.11"
     },
-    "minimum-stability": "dev",
+    "minimum-stability": "beta",
     "prefer-stable": true,
     "autoload": {
         "psr-4": {
diff --git a/src/DefaultPostgresConnector.php b/src/DefaultPostgresConnector.php
index adc2e40..569cad3 100644
--- a/src/DefaultPostgresConnector.php
+++ b/src/DefaultPostgresConnector.php
@@ -3,12 +3,13 @@
 namespace Amp\Postgres;
 
 use Amp\Cancellation;
+use Amp\Postgres\Internal\PostgresHandleConnection;
 use Amp\Sql\SqlConfig;
 use Amp\Sql\SqlConnector;
 use Amp\Sql\SqlException;
 
 /**
- * @implements SqlConnector<PostgresConfig, PostgresConnection>
+ * @implements SqlConnector<PostgresConfig, PostgresHandleConnection>
  */
 final class DefaultPostgresConnector implements SqlConnector
 {
@@ -17,7 +18,7 @@ final class DefaultPostgresConnector implements SqlConnector
      *
      * @throws \Error If neither ext-pgsql nor pecl-pq is loaded.
      */
-    public function connect(SqlConfig $config, ?Cancellation $cancellation = null): PostgresConnection
+    public function connect(SqlConfig $config, ?Cancellation $cancellation = null): PostgresHandleConnection
     {
         if (!$config instanceof PostgresConfig) {
             throw new \TypeError(\sprintf("Must provide an instance of %s to Postgres connectors", PostgresConfig::class));
diff --git a/src/Internal/AbstractHandle.php b/src/Internal/AbstractHandle.php
index abc6d3c..ed1d499 100644
--- a/src/Internal/AbstractHandle.php
+++ b/src/Internal/AbstractHandle.php
@@ -5,7 +5,7 @@
 use Amp\DeferredFuture;
 use Amp\Pipeline\Queue;
 use Amp\Postgres\ByteA;
-use Amp\Postgres\PostgresHandle;
+use Amp\Postgres\PostgresConfig;
 use Amp\Sql\ConnectionException;
 use Revolt\EventLoop;
 
@@ -22,6 +22,7 @@ abstract class AbstractHandle implements PostgresHandle
     protected int $lastUsedAt = 0;
 
     public function __construct(
+        private readonly PostgresConfig $config,
         protected readonly string $poll,
         protected readonly string $await,
         private readonly DeferredFuture $onClose,
@@ -36,6 +37,11 @@ public function __destruct()
         }
     }
 
+    public function getConfig(): PostgresConfig
+    {
+        return $this->config;
+    }
+
     public function getLastUsedAt(): int
     {
         return $this->lastUsedAt;
diff --git a/src/Internal/PgSqlHandle.php b/src/Internal/PgSqlHandle.php
index 060b432..eb002a4 100644
--- a/src/Internal/PgSqlHandle.php
+++ b/src/Internal/PgSqlHandle.php
@@ -5,7 +5,7 @@
 use Amp\DeferredFuture;
 use Amp\Future;
 use Amp\Pipeline\Queue;
-use Amp\Postgres\PostgresHandle;
+use Amp\Postgres\PostgresConfig;
 use Amp\Postgres\PostgresListener;
 use Amp\Postgres\PostgresNotification;
 use Amp\Postgres\PostgresResult;
@@ -54,8 +54,12 @@ final class PgSqlHandle extends AbstractHandle
      * @param resource $socket PostgreSQL connection stream socket.
      * @param string $id Connection identifier for determining which cached type table to use.
      */
-    public function __construct(\PgSql\Connection $handle, $socket, string $id)
-    {
+    public function __construct(
+        \PgSql\Connection $handle,
+        $socket,
+        string $id,
+        PostgresConfig $config,
+    ) {
         $this->handle = $handle;
 
         $handle = &$this->handle;
@@ -151,7 +155,7 @@ public function __construct(\PgSql\Connection $handle, $socket, string $id)
         EventLoop::unreference($poll);
         EventLoop::disable($await);
 
-        parent::__construct($poll, $await, $onClose);
+        parent::__construct($config, $poll, $await, $onClose);
 
         $this->types = (self::$typeCache[$id] ??= self::fetchTypes($handle));
     }
diff --git a/src/Internal/PostgresConnectionStatement.php b/src/Internal/PostgresConnectionStatement.php
index c3dae9f..aa0e470 100644
--- a/src/Internal/PostgresConnectionStatement.php
+++ b/src/Internal/PostgresConnectionStatement.php
@@ -3,7 +3,6 @@
 namespace Amp\Postgres\Internal;
 
 use Amp\DeferredFuture;
-use Amp\Postgres\PostgresHandle;
 use Amp\Postgres\PostgresStatement;
 use Amp\Sql\Result;
 use Amp\Sql\SqlException;
diff --git a/src/Internal/PostgresConnectionTransaction.php b/src/Internal/PostgresConnectionTransaction.php
index cd19895..08a244b 100644
--- a/src/Internal/PostgresConnectionTransaction.php
+++ b/src/Internal/PostgresConnectionTransaction.php
@@ -3,7 +3,6 @@
 namespace Amp\Postgres\Internal;
 
 use Amp\DeferredFuture;
-use Amp\Postgres\PostgresHandle;
 use Amp\Postgres\PostgresResult;
 use Amp\Postgres\PostgresStatement;
 use Amp\Postgres\PostgresTransaction;
@@ -13,7 +12,7 @@
 use Revolt\EventLoop;
 
 /** @internal  */
-final class PostgresConnectionTransaction implements PostgresTransaction
+final class PostgresConnectionTransaction implements PostgresNestableTransaction
 {
     /** @var \Closure():void */
     private readonly \Closure $release;
@@ -31,7 +30,7 @@ final class PostgresConnectionTransaction implements PostgresTransaction
     public function __construct(
         private readonly PostgresHandle $handle,
         \Closure $release,
-        private readonly TransactionIsolation $isolation
+        private readonly TransactionIsolation $isolation,
     ) {
         $refCount = &$this->refCount;
         $statements = &$this->statements;
@@ -73,6 +72,11 @@ public function getLastUsedAt(): int
         return $this->handle->getLastUsedAt();
     }
 
+    public function isNestedTransaction(): bool
+    {
+        return false;
+    }
+
     private function assertOpen(): void
     {
         if ($this->isClosed()) {
@@ -169,6 +173,22 @@ public function execute(string $sql, array $params = []): PostgresResult
         return new PostgresPooledResult($result, $this->release);
     }
 
+    public function beginTransaction(): PostgresTransaction
+    {
+        $this->assertOpen();
+
+        ++$this->refCount;
+        try {
+            $identifier = bin2hex(\random_bytes(4));
+            $this->createSavepoint($identifier);
+        } catch (\Throwable $exception) {
+            EventLoop::queue($this->release);
+            throw $exception;
+        }
+
+        return new PostgresNestedTransaction($this, $identifier, $this->release);
+    }
+
     /**
      * @throws TransactionError If the transaction has been committed or rolled back.
      */
@@ -238,6 +258,16 @@ public function releaseSavepoint(string $identifier): void
         $this->query("RELEASE SAVEPOINT " . $this->quoteName($identifier));
     }
 
+    public function onCommit(\Closure $onCommit): void
+    {
+        // TODO: Implement onCommit() method.
+    }
+
+    public function onRollback(\Closure $onRollback): void
+    {
+        // TODO: Implement onRollback() method.
+    }
+
     /**
      * @throws TransactionError If the transaction has been committed or rolled back.
      */
diff --git a/src/Internal/PostgresHandle.php b/src/Internal/PostgresHandle.php
new file mode 100644
index 0000000..e78bea1
--- /dev/null
+++ b/src/Internal/PostgresHandle.php
@@ -0,0 +1,46 @@
+<?php declare(strict_types=1);
+
+namespace Amp\Postgres\Internal;
+
+use Amp\Postgres\PostgresConfig;
+use Amp\Postgres\PostgresListener;
+use Amp\Postgres\PostgresQuoter;
+use Amp\Postgres\PostgresResult;
+use Amp\Postgres\PostgresStatement;
+use Amp\Sql\TransientResource;
+
+/** @internal */
+interface PostgresHandle extends PostgresQuoter, TransientResource
+{
+    public const STATEMENT_NAME_PREFIX = "amp_";
+
+    public function getConfig(): PostgresConfig;
+
+    public function query(string $sql): PostgresResult;
+
+    public function execute(string $sql, array $params = []): PostgresResult;
+
+    public function prepare(string $sql): PostgresStatement;
+
+    /**
+     * @param non-empty-string $channel
+     */
+    public function notify(string $channel, string $payload = ""): PostgresResult;
+
+    /**
+     * @param non-empty-string $channel
+     */
+    public function listen(string $channel): PostgresListener;
+
+    /**
+     * Execute the statement with the given name and parameters.
+     *
+     * @param list<mixed> $params List of statement parameters, indexed starting at 0.
+     */
+    public function statementExecute(string $name, array $params): PostgresResult;
+
+    /**
+     * Deallocate the statement with the given name.
+     */
+    public function statementDeallocate(string $name): void;
+}
diff --git a/src/Internal/PostgresHandleConnection.php b/src/Internal/PostgresHandleConnection.php
new file mode 100644
index 0000000..137846e
--- /dev/null
+++ b/src/Internal/PostgresHandleConnection.php
@@ -0,0 +1,157 @@
+<?php declare(strict_types=1);
+
+namespace Amp\Postgres\Internal;
+
+use Amp\Cancellation;
+use Amp\DeferredFuture;
+use Amp\Postgres\Internal;
+use Amp\Postgres\PostgresConfig;
+use Amp\Postgres\PostgresConnection;
+use Amp\Postgres\PostgresListener;
+use Amp\Postgres\PostgresResult;
+use Amp\Postgres\PostgresStatement;
+use Amp\Postgres\PostgresTransaction;
+use Amp\Sql\ConnectionException;
+use Amp\Sql\TransactionIsolation;
+use Amp\Sql\TransactionIsolationLevel;
+
+/** @internal */
+abstract class PostgresHandleConnection implements PostgresConnection
+{
+    /** @var DeferredFuture|null Used to only allow one transaction at a time. */
+    private ?DeferredFuture $busy = null;
+
+    private TransactionIsolation $transactionIsolation = TransactionIsolationLevel::Committed;
+
+    /**
+     * @throws ConnectionException
+     */
+    abstract public static function connect(
+        PostgresConfig $config,
+        ?Cancellation $cancellation = null,
+    ): self;
+
+    protected function __construct(private readonly PostgresHandle $handle)
+    {
+    }
+
+    final public function getConfig(): PostgresConfig
+    {
+        return $this->handle->getConfig();
+    }
+
+    final public function getLastUsedAt(): int
+    {
+        return $this->handle->getLastUsedAt();
+    }
+
+    final public function close(): void
+    {
+        $this->handle->close();
+    }
+
+    final public function isClosed(): bool
+    {
+        return $this->handle->isClosed();
+    }
+
+    final public function onClose(\Closure $onClose): void
+    {
+        $this->handle->onClose($onClose);
+    }
+
+    private function awaitPending(): void
+    {
+        while ($this->busy) {
+            $this->busy->getFuture()->await();
+        }
+    }
+
+    /**
+     * Reserves the connection for a transaction.
+     */
+    private function reserve(): void
+    {
+        \assert($this->busy === null);
+        $this->busy = new DeferredFuture;
+    }
+
+    /**
+     * Releases the transaction lock.
+     */
+    private function release(): void
+    {
+        \assert($this->busy !== null);
+
+        $this->busy->complete();
+        $this->busy = null;
+    }
+
+    final public function query(string $sql): PostgresResult
+    {
+        $this->awaitPending();
+        return $this->handle->query($sql);
+    }
+
+    final public function execute(string $sql, array $params = []): PostgresResult
+    {
+        $this->awaitPending();
+        return $this->handle->execute($sql, $params);
+    }
+
+    final public function prepare(string $sql): PostgresStatement
+    {
+        $this->awaitPending();
+        return $this->handle->prepare($sql);
+    }
+
+    final public function notify(string $channel, string $payload = ""): PostgresResult
+    {
+        $this->awaitPending();
+        return $this->handle->notify($channel, $payload);
+    }
+
+    final public function listen(string $channel): PostgresListener
+    {
+        $this->awaitPending();
+        return $this->handle->listen($channel);
+    }
+
+    final public function beginTransaction(): PostgresTransaction
+    {
+        $this->reserve();
+
+        try {
+            $this->handle->query("BEGIN TRANSACTION ISOLATION LEVEL " . $this->transactionIsolation->toSql());
+        } catch (\Throwable $exception) {
+            $this->release();
+            throw $exception;
+        }
+
+        return new Internal\PostgresConnectionTransaction(
+            $this->handle,
+            $this->release(...),
+            $this->transactionIsolation,
+        );
+    }
+
+    final public function setTransactionIsolation(TransactionIsolation $isolation): void
+    {
+        $this->transactionIsolation = $isolation;
+    }
+
+    final public function quoteString(string $data): string
+    {
+        return $this->handle->quoteString($data);
+    }
+
+    final public function quoteName(string $name): string
+    {
+        return $this->handle->quoteName($name);
+    }
+
+    final public function escapeByteA(string $data): string
+    {
+        return $this->handle->escapeByteA($data);
+    }
+}
diff --git a/src/Internal/PostgresNestableTransaction.php b/src/Internal/PostgresNestableTransaction.php
new file mode 100644
index 0000000..64596ec
--- /dev/null
+++ b/src/Internal/PostgresNestableTransaction.php
@@ -0,0 +1,16 @@
+<?php
+
+namespace Amp\Postgres\Internal;
+
+use Amp\Postgres\PostgresResult;
+use Amp\Postgres\PostgresStatement;
+use Amp\Postgres\PostgresTransaction;
+use Amp\Sql\Common\NestableTransaction;
+
+/**
+ * @internal
+ * @extends NestableTransaction<PostgresResult, PostgresStatement, PostgresTransaction>
+ */
+interface PostgresNestableTransaction extends NestableTransaction, PostgresTransaction
+{
+}
diff --git a/src/Internal/PostgresNestedTransaction.php b/src/Internal/PostgresNestedTransaction.php
index 26d807f..3218d76 100644
--- a/src/Internal/PostgresNestedTransaction.php
+++ b/src/Internal/PostgresNestedTransaction.php
@@ -1,17 +1,19 @@
-<?php declare(strict_types=1);
+<?php
 
 namespace Amp\Postgres\Internal;
 
 use Amp\Postgres\PostgresResult;
 use Amp\Postgres\PostgresStatement;
 use Amp\Postgres\PostgresTransaction;
+use Amp\Sql\Common\NestableTransaction;
 use Amp\Sql\Common\NestedTransaction;
+use Amp\Sql\Transaction;
 
 /**
  * @internal
- * @extends NestedTransaction<PostgresResult, PostgresStatement, PostgresTransaction>
+ * @extends NestedTransaction<PostgresResult, PostgresStatement, PostgresTransaction, PostgresNestableTransaction>
  */
-final class PostgresNestedTransaction extends NestedTransaction implements PostgresTransaction
+final class PostgresNestedTransaction extends NestedTransaction implements PostgresNestableTransaction
 {
     use PostgresTransactionDelegate;
 
@@ -19,4 +21,12 @@ protected function getTransaction(): PostgresTransaction
     {
         return $this->transaction;
     }
+
+    protected function createNestedTransaction(
+        NestableTransaction $transaction,
+        string $identifier,
+        \Closure $release,
+    ): Transaction {
+        return new PostgresNestedTransaction($transaction, $identifier, $release);
+    }
 }
diff --git a/src/Internal/PostgresPooledTransaction.php b/src/Internal/PostgresPooledTransaction.php
index cfa9f66..ef4eac6 100644
--- a/src/Internal/PostgresPooledTransaction.php
+++ b/src/Internal/PostgresPooledTransaction.php
@@ -6,6 +6,7 @@
 use Amp\Postgres\PostgresStatement;
 use Amp\Postgres\PostgresTransaction;
 use Amp\Sql\Common\PooledTransaction;
+use Amp\Sql\Transaction;
 
 /**
  * @internal
@@ -19,4 +20,10 @@ protected function getTransaction(): PostgresTransaction
     {
         return $this->transaction;
     }
+
+    protected function createTransaction(Transaction $transaction, \Closure $release): PostgresTransaction
+    {
+        \assert($transaction instanceof PostgresTransaction);
+        return new PostgresPooledTransaction($transaction, $release);
+    }
 }
diff --git a/src/Internal/PostgresStatementPool.php b/src/Internal/PostgresStatementPool.php
index 97767f7..fd4c68c 100644
--- a/src/Internal/PostgresStatementPool.php
+++ b/src/Internal/PostgresStatementPool.php
@@ -2,6 +2,7 @@
 
 namespace Amp\Postgres\Internal;
 
+use Amp\Postgres\PostgresConfig;
 use Amp\Postgres\PostgresResult;
 use Amp\Postgres\PostgresStatement;
 use Amp\Postgres\PostgresTransaction;
@@ -10,7 +11,7 @@
 
 /**
  * @internal
- * @extends SqlStatementPool<PostgresResult, PostgresStatement, PostgresTransaction>
+ * @extends SqlStatementPool<PostgresConfig, PostgresResult, PostgresStatement, PostgresTransaction>
  */
 final class PostgresStatementPool extends SqlStatementPool implements PostgresStatement
 {
diff --git a/src/Internal/PostgresTransactionDelegate.php b/src/Internal/PostgresTransactionDelegate.php
index edb310f..9daaaa0 100644
--- a/src/Internal/PostgresTransactionDelegate.php
+++ b/src/Internal/PostgresTransactionDelegate.php
@@ -55,6 +55,14 @@ public function execute(string $sql, array $params = []): PostgresResult
         return parent::execute($sql, $params);
     }
 
+    /**
+     * Changes return type to this library's Transaction type.
+     */
+    public function beginTransaction(): PostgresTransaction
+    {
+        return parent::beginTransaction();
+    }
+
     /**
      * @param non-empty-string $channel
      */
diff --git a/src/Internal/PqHandle.php b/src/Internal/PqHandle.php
index 3694e4e..a366ad2 100644
--- a/src/Internal/PqHandle.php
+++ b/src/Internal/PqHandle.php
@@ -5,7 +5,7 @@
 use Amp\DeferredFuture;
 use Amp\Future;
 use Amp\Pipeline\Queue;
-use Amp\Postgres\PostgresHandle;
+use Amp\Postgres\PostgresConfig;
 use Amp\Postgres\PostgresListener;
 use Amp\Postgres\PostgresNotification;
 use Amp\Postgres\PostgresResult;
@@ -28,7 +28,7 @@ final class PqHandle extends AbstractHandle
     /** @var array<non-empty-string, StatementStorage<pq\Statement>> */
     private array $statements = [];
 
-    public function __construct(pq\Connection $handle)
+    public function __construct(pq\Connection $handle, PostgresConfig $config)
     {
         $this->handle = $handle;
 
@@ -103,7 +103,7 @@ public function __construct(pq\Connection $handle)
         EventLoop::unreference($poll);
         EventLoop::disable($await);
 
-        parent::__construct($poll, $await, $onClose);
+        parent::__construct($config, $poll, $await, $onClose);
     }
 
     public function isClosed(): bool
diff --git a/src/PgSqlConnection.php b/src/PgSqlConnection.php
index 0ddc91c..c15023b 100644
--- a/src/PgSqlConnection.php
+++ b/src/PgSqlConnection.php
@@ -7,12 +7,12 @@
 use Amp\Sql\ConnectionException;
 use Revolt\EventLoop;
 
-final class PgSqlConnection extends PostgresConnection implements PostgresLink
+final class PgSqlConnection extends Internal\PostgresHandleConnection implements PostgresConnection
 {
     /**
      * @throws \Error If pecl-ev is used as a loop extension.
      */
-    public static function connect(PostgresConfig $connectionConfig, ?Cancellation $cancellation = null): self
+    public static function connect(PostgresConfig $config, ?Cancellation $cancellation = null): self
     {
         // @codeCoverageIgnoreStart
         /** @psalm-suppress UndefinedClass */
@@ -20,7 +20,7 @@ public static function connect(PostgresConfig $connectionConfig, ?Cancellation $
             throw new \Error('ext-pgsql is not compatible with pecl-ev; use pecl-pq or a different loop extension');
         } // @codeCoverageIgnoreEnd
 
-        if (!$connection = \pg_connect($connectionConfig->getConnectionString(), \PGSQL_CONNECT_ASYNC | \PGSQL_CONNECT_FORCE_NEW)) {
+        if (!$connection = \pg_connect($config->getConnectionString(), \PGSQL_CONNECT_ASYNC | \PGSQL_CONNECT_FORCE_NEW)) {
             throw new ConnectionException("Failed to create connection resource");
         }
 
@@ -32,11 +32,11 @@ public static function connect(PostgresConfig $connectionConfig, ?Cancellation $
             throw new ConnectionException("Failed to access connection socket");
         }
 
-        $hash = \sha1($connectionConfig->getHost() . $connectionConfig->getPort() . $connectionConfig->getUser());
+        $hash = \sha1($config->getHost() . $config->getPort() . $config->getUser());
 
         $deferred = new DeferredFuture();
         /** @psalm-suppress MissingClosureParamType $resource is a resource and cannot be inferred in this context */
-        $callback = static function (string $callbackId, $resource) use (&$poll, &$await, $connection, $deferred, $hash): void {
+        $callback = static function (string $callbackId, $resource) use (&$poll, &$await, $connection, $config, $deferred, $hash): void {
             switch ($result = \pg_connect_poll($connection)) {
                 case \PGSQL_POLLING_READING:
                 case \PGSQL_POLLING_WRITING:
@@ -47,7 +47,7 @@ public static function connect(PostgresConfig $connectionConfig, ?Cancellation $
                     break;
 
                 case \PGSQL_POLLING_OK:
-                    $deferred->complete(new self($connection, $resource, $hash));
+                    $deferred->complete(new self($connection, $resource, $hash, $config));
                     break;
 
                 default:
@@ -75,8 +75,8 @@ public static function connect(PostgresConfig $connectionConfig, ?Cancellation $
      * @param resource $socket PostgreSQL connection stream socket.
      * @param string $id Connection identifier for determining which cached type table to use.
      */
-    protected function __construct(\PgSql\Connection $handle, $socket, string $id)
+    protected function __construct(\PgSql\Connection $handle, $socket, string $id, PostgresConfig $config)
     {
-        parent::__construct(new Internal\PgSqlHandle($handle, $socket, $id));
+        parent::__construct(new Internal\PgSqlHandle($handle, $socket, $id, $config));
     }
 }
diff --git a/src/PostgresConnection.php b/src/PostgresConnection.php
index 16557c6..b23722f 100644
--- a/src/PostgresConnection.php
+++ b/src/PostgresConnection.php
@@ -2,133 +2,15 @@
 
 namespace Amp\Postgres;
 
-use Amp\Cancellation;
-use Amp\DeferredFuture;
-use Amp\Sql\ConnectionException;
-use Amp\Sql\TransactionIsolation;
-use Amp\Sql\TransactionIsolationLevel;
+use Amp\Sql\Connection;
 
-abstract class PostgresConnection implements PostgresLink, PostgresReceiver
+/**
+ * @extends Connection<PostgresConfig, PostgresResult, PostgresStatement, PostgresTransaction>
+ */
+interface PostgresConnection extends Connection, PostgresLink
 {
-    /** @var DeferredFuture|null Used to only allow one transaction at a time. */
-    private ?DeferredFuture $busy = null;
-
-    /**
-     * @throws ConnectionException
-     */
-    abstract public static function connect(
-        PostgresConfig $connectionConfig,
-        ?Cancellation $cancellation = null,
-    ): self;
-
-    protected function __construct(private readonly PostgresHandle $handle)
-    {
-    }
-
-    final public function getLastUsedAt(): int
-    {
-        return $this->handle->getLastUsedAt();
-    }
-
-    final public function close(): void
-    {
-        $this->handle->close();
-    }
-
-    final public function isClosed(): bool
-    {
-        return $this->handle->isClosed();
-    }
-
-    final public function onClose(\Closure $onClose): void
-    {
-        $this->handle->onClose($onClose);
-    }
-
-    private function awaitPending(): void
-    {
-        while ($this->busy) {
-            $this->busy->getFuture()->await();
-        }
-    }
-
-    /**
-     * Reserves the connection for a transaction.
-     */
-    private function reserve(): void
-    {
-        \assert($this->busy === null);
-        $this->busy = new DeferredFuture;
-    }
-
     /**
-     * Releases the transaction lock.
+     * @return PostgresConfig Config object specific to this library.
      */
-    private function release(): void
-    {
-        \assert($this->busy !== null);
-
-        $this->busy->complete();
-        $this->busy = null;
-    }
-
-    final public function query(string $sql): PostgresResult
-    {
-        $this->awaitPending();
-        return $this->handle->query($sql);
-    }
-
-    final public function execute(string $sql, array $params = []): PostgresResult
-    {
-        $this->awaitPending();
-        return $this->handle->execute($sql, $params);
-    }
-
-    final public function prepare(string $sql): PostgresStatement
-    {
-        $this->awaitPending();
-        return $this->handle->prepare($sql);
-    }
-
-    final public function notify(string $channel, string $payload = ""): PostgresResult
-    {
-        $this->awaitPending();
-        return $this->handle->notify($channel, $payload);
-    }
-
-    final public function listen(string $channel): PostgresListener
-    {
-        $this->awaitPending();
-        return $this->handle->listen($channel);
-    }
-
-    final public function beginTransaction(
-        TransactionIsolation $isolation = TransactionIsolationLevel::Committed
-    ): PostgresTransaction {
-        $this->reserve();
-
-        try {
-            $this->handle->query("BEGIN TRANSACTION ISOLATION LEVEL " . $isolation->toSql());
-        } catch (\Throwable $exception) {
-            $this->release();
-            throw $exception;
-        }
-
-        return new Internal\PostgresConnectionTransaction($this->handle, $this->release(...), $isolation);
-    }
-
-    final public function quoteString(string $data): string
-    {
-        return $this->handle->quoteString($data);
-    }
-
-    final public function quoteName(string $name): string
-    {
-        return $this->handle->quoteName($name);
-    }
-
-    final public function escapeByteA(string $data): string
-    {
-        return $this->handle->escapeByteA($data);
-    }
+    public function getConfig(): PostgresConfig;
 }
diff --git a/src/PostgresConnectionPool.php b/src/PostgresConnectionPool.php
index e92d0f8..cdad2e5 100644
--- a/src/PostgresConnectionPool.php
+++ b/src/PostgresConnectionPool.php
@@ -3,21 +3,20 @@
 namespace Amp\Postgres;
 
 use Amp\Future;
+use Amp\Postgres\Internal\PostgresHandleConnection;
 use Amp\Sql\Common\ConnectionPool;
 use Amp\Sql\Result;
 use Amp\Sql\SqlConnector;
 use Amp\Sql\Statement;
 use Amp\Sql\Transaction;
-use Amp\Sql\TransactionIsolation;
-use Amp\Sql\TransactionIsolationLevel;
 use function Amp\async;
 
 /**
- * @extends ConnectionPool<PostgresConfig, PostgresResult, PostgresStatement, PostgresTransaction, PostgresConnection>
+ * @extends ConnectionPool<PostgresConfig, PostgresResult, PostgresStatement, PostgresTransaction, PostgresHandleConnection>
  */
-final class PostgresConnectionPool extends ConnectionPool implements PostgresLink, PostgresReceiver
+final class PostgresConnectionPool extends ConnectionPool implements PostgresLink
 {
-    /** @var Future<PostgresConnection>|null Connection used for notification listening. */
+    /** @var Future<PostgresHandleConnection>|null Connection used for notification listening. */
     private Future|null $listeningConnection = null;
 
     /** @var int Number of listeners on listening connection. */
@@ -27,7 +26,7 @@ final class PostgresConnectionPool extends ConnectionPool implements PostgresLin
      * @param positive-int $maxConnections
      * @param positive-int $idleTimeout
      * @param bool $resetConnections True to automatically execute DISCARD ALL on a connection before use.
-     * @param SqlConnector<PostgresConfig, PostgresConnection>|null $connector
+     * @param SqlConnector<PostgresConfig, PostgresHandleConnection>|null $connector
      */
     public function __construct(
         PostgresConfig $config,
@@ -68,13 +67,13 @@ protected function createTransaction(Transaction $transaction, \Closure $release
     /**
      * Changes return type to this library's Transaction type.
      */
-    public function beginTransaction(
-        TransactionIsolation $isolation = TransactionIsolationLevel::Committed
-    ): PostgresTransaction {
-        return parent::beginTransaction($isolation);
+
+    public function beginTransaction(): PostgresTransaction
+    {
+        return parent::beginTransaction();
     }
 
-    protected function pop(): PostgresConnection
+    protected function pop(): PostgresHandleConnection
     {
         $connection = parent::pop();
 
diff --git a/src/PostgresExecutor.php b/src/PostgresExecutor.php
index 96b9652..9604690 100644
--- a/src/PostgresExecutor.php
+++ b/src/PostgresExecutor.php
@@ -7,24 +7,29 @@
 use Amp\Sql\SqlException;
 
 /**
- * @extends Executor<PostgresResult, PostgresStatement>
+ * @extends Executor<PostgresResult, PostgresStatement, PostgresTransaction>
  */
 interface PostgresExecutor extends Executor, PostgresQuoter
 {
-    /**
-     * @return PostgresResult Result object specific to this library.
-     */
-    public function query(string $sql): PostgresResult;
-
-    /**
-     * @return PostgresStatement Statement object specific to this library.
-     */
-    public function prepare(string $sql): PostgresStatement;
-
-    /**
-     * @return PostgresResult Result object specific to this library.
-     */
-    public function execute(string $sql, array $params = []): PostgresResult;
+//    /**
+//     * @return PostgresResult Result object specific to this library.
+//     */
+//    public function query(string $sql): PostgresResult;
+//
+//    /**
+//     * @return PostgresStatement Statement object specific to this library.
+//     */
+//    public function prepare(string $sql): PostgresStatement;
+//
+//    /**
+//     * @return PostgresResult Result object specific to this library.
+//     */
+//    public function execute(string $sql, array $params = []): PostgresResult;
+//
+//    /**
+//     * @return PostgresTransaction Transaction object specific to this library.
+//     */
+//    public function beginTransaction(): PostgresTransaction;
 
     /**
      * @param non-empty-string $channel Channel name.
diff --git a/src/PostgresHandle.php b/src/PostgresHandle.php
deleted file mode 100644
index 79ea9a0..0000000
--- a/src/PostgresHandle.php
+++ /dev/null
@@ -1,20 +0,0 @@
-<?php declare(strict_types=1);
-
-namespace Amp\Postgres;
-
-interface PostgresHandle extends PostgresReceiver
-{
-    public const STATEMENT_NAME_PREFIX = "amp_";
-
-    /**
-     * Execute the statement with the given name and parameters.
-     *
-     * @param list<mixed> $params List of statement parameters, indexed starting at 0.
-     */
-    public function statementExecute(string $name, array $params): PostgresResult;
-
-    /**
-     * Deallocate the statement with the given name.
-     */
-    public function statementDeallocate(string $name): void;
-}
diff --git a/src/PostgresLink.php b/src/PostgresLink.php
index de5a08f..4c084f1 100644
--- a/src/PostgresLink.php
+++ b/src/PostgresLink.php
@@ -2,19 +2,22 @@
 
 namespace Amp\Postgres;
 
+use Amp\Sql\ConnectionException;
 use Amp\Sql\Link;
-use Amp\Sql\TransactionIsolation;
-use Amp\Sql\TransactionIsolationLevel;
+use Amp\Sql\QueryError;
+use Amp\Sql\SqlException;
 
 /**
  * @extends Link<PostgresResult, PostgresStatement, PostgresTransaction>
  */
-interface PostgresLink extends Link, PostgresQuoter
+interface PostgresLink extends Link, PostgresExecutor
 {
     /**
-     * @return PostgresTransaction Transaction object specific to this library.
+     * @param non-empty-string $channel Channel name.
+     *
+     * @throws SqlException If the operation fails due to unexpected condition.
+     * @throws ConnectionException If the connection to the database is lost.
+     * @throws QueryError If the operation fails due to an error in the query (such as a syntax error).
      */
-    public function beginTransaction(
-        TransactionIsolation $isolation = TransactionIsolationLevel::Committed
-    ): PostgresTransaction;
+    public function listen(string $channel): PostgresListener;
 }
diff --git a/src/PostgresNestableTransaction.php b/src/PostgresNestableTransaction.php
deleted file mode 100644
index 05023f2..0000000
--- a/src/PostgresNestableTransaction.php
+++ /dev/null
@@ -1,46 +0,0 @@
-<?php declare(strict_types=1);
-
-namespace Amp\Postgres;
-
-use Amp\Sql\Common\NestableTransaction;
-use Amp\Sql\Transaction;
-use Amp\Sql\TransactionIsolation;
-use Amp\Sql\TransactionIsolationLevel;
-
-/**
- * @extends NestableTransaction<PostgresResult, PostgresStatement, PostgresTransaction>
- */
-final class PostgresNestableTransaction extends NestableTransaction implements PostgresLink
-{
-    protected function createNestedTransaction(
-        Transaction $transaction,
-        \Closure $release,
-        string $identifier,
-    ): Transaction {
-        return new Internal\PostgresNestedTransaction($transaction, $release, $identifier);
-    }
-
-    /**
-     * Changes return type to this library's Transaction type.
-     */
-    public function beginTransaction(
-        TransactionIsolation $isolation = TransactionIsolationLevel::Committed
-    ): PostgresTransaction {
-        return parent::beginTransaction($isolation);
-    }
-
-    public function quoteString(string $data): string
-    {
-        return $this->transaction->quoteString($data);
-    }
-
-    public function quoteName(string $name): string
-    {
-        return $this->transaction->quoteName($name);
-    }
-
-    public function escapeByteA(string $data): string
-    {
-        return $this->transaction->escapeByteA($data);
-    }
-}
diff --git a/src/PostgresReceiver.php b/src/PostgresReceiver.php
deleted file mode 100644
index e7c4803..0000000
--- a/src/PostgresReceiver.php
+++ /dev/null
@@ -1,19 +0,0 @@
-<?php declare(strict_types=1);
-
-namespace Amp\Postgres;
-
-use Amp\Sql\ConnectionException;
-use Amp\Sql\QueryError;
-use Amp\Sql\SqlException;
-
-interface PostgresReceiver extends PostgresExecutor
-{
-    /**
-     * @param non-empty-string $channel Channel name.
-     *
-     * @throws SqlException If the operation fails due to unexpected condition.
-     * @throws ConnectionException If the connection to the database is lost.
-     * @throws QueryError If the operation fails due to an error in the query (such as a syntax error).
-     */
-    public function listen(string $channel): PostgresListener;
-}
diff --git a/src/PostgresTransaction.php b/src/PostgresTransaction.php
index add8dc8..2f74100 100644
--- a/src/PostgresTransaction.php
+++ b/src/PostgresTransaction.php
@@ -7,8 +7,8 @@
 /**
  * Note that notifications sent during a transaction are not delivered until the transaction has been committed.
  *
- * @extends Transaction<PostgresResult, PostgresStatement>
+ * @extends Transaction<PostgresResult, PostgresStatement, PostgresTransaction>
  */
-interface PostgresTransaction extends PostgresExecutor, Transaction
+interface PostgresTransaction extends Transaction, PostgresExecutor
 {
 }
diff --git a/src/PqConnection.php b/src/PqConnection.php
index 77b1960..23a94c2 100644
--- a/src/PqConnection.php
+++ b/src/PqConnection.php
@@ -8,14 +8,14 @@
 use pq;
 use Revolt\EventLoop;
 
-final class PqConnection extends PostgresConnection implements PostgresLink
+final class PqConnection extends Internal\PostgresHandleConnection implements PostgresConnection
 {
     private readonly Internal\PqHandle $handle;
 
-    public static function connect(PostgresConfig $connectionConfig, ?Cancellation $cancellation = null): self
+    public static function connect(PostgresConfig $config, ?Cancellation $cancellation = null): self
     {
         try {
-            $connection = new pq\Connection($connectionConfig->getConnectionString(), pq\Connection::ASYNC);
+            $connection = new pq\Connection($config->getConnectionString(), pq\Connection::ASYNC);
         } catch (pq\Exception $exception) {
             throw new ConnectionException("Could not connect to PostgreSQL server", 0, $exception);
         }
@@ -24,7 +24,7 @@ public static function connect(PostgresConfig $connectionConfig, ?Cancellation $
         $connection->unbuffered = true;
 
         $deferred = new DeferredFuture();
-        $callback = function () use (&$poll, &$await, $connection, $deferred): void {
+        $callback = function () use (&$poll, &$await, $connection, $config, $deferred): void {
             switch ($result = $connection->poll()) {
                 case pq\Connection::POLLING_READING:
                 case pq\Connection::POLLING_WRITING:
@@ -35,7 +35,7 @@ public static function connect(PostgresConfig $connectionConfig, ?Cancellation $
                     break;
 
                 case pq\Connection::POLLING_OK:
-                    $deferred->complete(new self($connection));
+                    $deferred->complete(new self($connection, $config));
                     break;
 
                 default:
@@ -58,9 +58,9 @@ public static function connect(PostgresConfig $connectionConfig, ?Cancellation $
         }
     }
 
-    protected function __construct(pq\Connection $handle)
+    protected function __construct(pq\Connection $handle, PostgresConfig $config)
     {
-        $this->handle = new Internal\PqHandle($handle);
+        $this->handle = new Internal\PqHandle($handle, $config);
         parent::__construct($this->handle);
     }
 
diff --git a/src/functions.php b/src/functions.php
index c1c7886..9bf65ad 100644
--- a/src/functions.php
+++ b/src/functions.php
@@ -3,15 +3,16 @@
 namespace Amp\Postgres;
 
 use Amp\Cancellation;
+use Amp\Postgres\Internal\PostgresHandleConnection;
 use Amp\Sql\Common\RetrySqlConnector;
 use Amp\Sql\SqlConnector;
 use Amp\Sql\SqlException;
 use Revolt\EventLoop;
 
 /**
- * @param SqlConnector<PostgresConfig, PostgresConnection>|null $connector
+ * @param SqlConnector<PostgresConfig, PostgresHandleConnection>|null $connector
  *
- * @return SqlConnector<PostgresConfig, PostgresConnection>
+ * @return SqlConnector<PostgresConfig, PostgresHandleConnection>
  */
 function postgresConnector(?SqlConnector $connector = null): SqlConnector
 {
@@ -25,7 +26,7 @@ function postgresConnector(?SqlConnector $connector = null): SqlConnector
 
     /**
      * @psalm-suppress InvalidArgument
-     * @var SqlConnector<PostgresConfig, PostgresConnection>
+     * @var SqlConnector<PostgresConfig, PostgresHandleConnection>
      */
     return $map[$driver] ??= new RetrySqlConnector(new DefaultPostgresConnector());
 }
@@ -37,7 +38,7 @@ function postgresConnector(?SqlConnector $connector = null): SqlConnector
  *
  * @throws \Error If neither ext-pgsql or pecl-pq is loaded.
  */
-function connect(PostgresConfig $config, ?Cancellation $cancellation = null): PostgresConnection
+function connect(PostgresConfig $config, ?Cancellation $cancellation = null): PostgresHandleConnection
 {
     return postgresConnector()->connect($config, $cancellation);
 }
diff --git a/test/AbstractConnectTest.php b/test/AbstractConnectTest.php
index 3ea0fed..6e35b91 100644
--- a/test/AbstractConnectTest.php
+++ b/test/AbstractConnectTest.php
@@ -4,13 +4,13 @@
 
 use Amp\Cancellation;
 use Amp\PHPUnit\AsyncTestCase;
+use Amp\Postgres\Internal\PostgresHandleConnection;
 use Amp\Postgres\PostgresConfig;
-use Amp\Postgres\PostgresConnection;
 
 abstract class AbstractConnectTest extends AsyncTestCase
 {
     abstract public function connect(
         PostgresConfig $connectionConfig,
         ?Cancellation $cancellation = null
-    ): PostgresConnection;
+    ): PostgresHandleConnection;
 }
diff --git a/test/AbstractConnectionTest.php b/test/AbstractConnectionTest.php
index 3426966..b860310 100644
--- a/test/AbstractConnectionTest.php
+++ b/test/AbstractConnectionTest.php
@@ -16,13 +16,13 @@ abstract class AbstractConnectionTest extends AbstractLinkTest
 {
     public function testIsClosed()
     {
-        $this->assertFalse($this->link->isClosed());
+        $this->assertFalse($this->executor->isClosed());
     }
 
     public function testConnectionCloseDuringQuery(): void
     {
-        $query = async($this->link->execute(...), 'SELECT pg_sleep(10)');
-        $close = async($this->link->close(...));
+        $query = async($this->executor->execute(...), 'SELECT pg_sleep(10)');
+        $close = async($this->executor->close(...));
 
         $start = \microtime(true);
 
@@ -41,14 +41,14 @@ public function testConnectionCloseDuringQuery(): void
     public function testListen()
     {
         $channel = "test";
-        $listener = $this->link->listen($channel);
+        $listener = $this->executor->listen($channel);
 
         $this->assertInstanceOf(PostgresListener::class, $listener);
         $this->assertSame($channel, $listener->getChannel());
 
         EventLoop::delay(0.1, function () use ($channel): void {
-            $this->link->query(\sprintf("NOTIFY %s, '%s'", $channel, '0'));
-            $this->link->query(\sprintf("NOTIFY %s, '%s'", $channel, '1'));
+            $this->executor->query(\sprintf("NOTIFY %s, '%s'", $channel, '0'));
+            $this->executor->query(\sprintf("NOTIFY %s, '%s'", $channel, '1'));
         });
 
         $count = 0;
@@ -68,11 +68,11 @@ public function testListen()
     public function testNotify()
     {
         $channel = "test";
-        $listener = $this->link->listen($channel);
+        $listener = $this->executor->listen($channel);
 
         EventLoop::delay(0.1, function () use ($channel) {
-            $this->link->notify($channel, '0');
-            $this->link->notify($channel, '1');
+            $this->executor->notify($channel, '0');
+            $this->executor->notify($channel, '1');
         });
 
         $count = 0;
@@ -95,18 +95,18 @@ public function testListenOnSameChannel()
         $this->expectExceptionMessage('Already listening on channel');
 
         $channel = "test";
-        Future\await([$this->link->listen($channel), $this->link->listen($channel)]);
+        Future\await([$this->executor->listen($channel), $this->executor->listen($channel)]);
     }
 
     public function testQueryAfterErroredQuery()
     {
         try {
-            $result = $this->link->query("INSERT INTO test VALUES ('github', 'com', '{1, 2, 3}', true, 4.2)");
+            $result = $this->executor->query("INSERT INTO test VALUES ('github', 'com', '{1, 2, 3}', true, 4.2)");
         } catch (QueryExecutionError $exception) {
             // Expected exception due to duplicate key.
         }
 
-        $result = $this->link->query("INSERT INTO test VALUES ('gitlab', 'com', '{1, 2, 3}', true, 4.2)");
+        $result = $this->executor->query("INSERT INTO test VALUES ('gitlab', 'com', '{1, 2, 3}', true, 4.2)");
 
         $this->assertSame(1, $result->getRowCount());
     }
diff --git a/test/AbstractLinkTest.php b/test/AbstractLinkTest.php
index b897597..f48657b 100644
--- a/test/AbstractLinkTest.php
+++ b/test/AbstractLinkTest.php
@@ -5,15 +5,14 @@
 use Amp\Future;
 use Amp\PHPUnit\AsyncTestCase;
 use Amp\Postgres\ByteA;
-use Amp\Postgres\PostgresConnection;
-use Amp\Postgres\PostgresLink;
+use Amp\Postgres\Internal\PostgresHandleConnection;
+use Amp\Postgres\PostgresExecutor;
 use Amp\Postgres\PostgresTransaction;
 use Amp\Postgres\QueryExecutionError;
 use Amp\Sql\QueryError;
 use Amp\Sql\Result;
 use Amp\Sql\Statement;
 use Amp\Sql\TransactionError;
-use Amp\Sql\TransactionIsolationLevel;
 use function Amp\async;
 
 abstract class AbstractLinkTest extends AsyncTestCase
@@ -31,7 +30,7 @@ abstract class AbstractLinkTest extends AsyncTestCase
     protected const INSERT_QUERY = 'INSERT INTO test VALUES ($1, $2, $3, $4, $5, $6, $7, $8)';
     protected const FIELD_COUNT = 8;
 
-    protected PostgresLink $link;
+    protected PostgresExecutor $executor;
 
     private ?array $data = null;
 
@@ -74,12 +73,12 @@ protected function verifyResult(Result $result, array $data): void
     }
 
     /**
-     * @return PostgresLink Connection or Link object to be tested.
+     * @return PostgresExecutor Executor object to be tested.
      */
-    abstract public function createLink(string $connectionString): PostgresLink;
+    abstract public function createExecutor(string $connectionString): PostgresExecutor;
 
     /**
-     * Helper method to invoke the protected constructor of classes extending {@see PostgresConnection}.
+     * Helper method to invoke the protected constructor of classes extending {@see PostgresHandleConnection}.
      *
      * @template T extends Connection
      *
@@ -88,7 +87,7 @@ abstract public function createLink(string $connectionString): PostgresLink;
      *
      * @return T
      */
-    protected function newConnection(string $className, mixed ...$args): PostgresConnection
+    protected function newConnection(string $className, mixed ...$args): PostgresHandleConnection
     {
         $reflection = new \ReflectionClass($className);
         $connection = $reflection->newInstanceWithoutConstructor();
@@ -99,18 +98,18 @@ protected function newConnection(string $className, mixed ...$args): PostgresCon
     public function setUp(): void
     {
         parent::setUp();
-        $this->link = $this->createLink('host=localhost user=postgres password=postgres');
+        $this->executor = $this->createExecutor('host=localhost user=postgres password=postgres');
     }
 
     public function tearDown(): void
     {
         parent::tearDown();
-        $this->link->close();
+        $this->executor->close();
     }
 
     public function testQueryFetchRow(): void
     {
-        $result = $this->link->query("SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test");
 
         $data = $this->getData();
         while ($row = $result->fetchRow()) {
@@ -122,7 +121,7 @@ public function testQueryFetchRow(): void
 
     public function testQueryWithTupleResult()
     {
-        $result = $this->link->query("SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test");
 
         $data = $this->getData();
 
@@ -133,7 +132,7 @@ public function testQueryWithTupleResult()
 
     public function testMultipleQueryWithTupleResult()
     {
-        $result = $this->link->query("SELECT * FROM test; SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test; SELECT * FROM test");
 
         $data = $this->getData();
 
@@ -150,7 +149,7 @@ public function testMultipleQueryWithTupleResult()
 
     public function testMultipleQueryWithCommandResultFirst()
     {
-        $result = $this->link->query("INSERT INTO test VALUES ('canon', 'jp', '{1}', true, 4.2, null, null, '3.1415926'); SELECT * FROM test");
+        $result = $this->executor->query("INSERT INTO test VALUES ('canon', 'jp', '{1}', true, 4.2, null, null, '3.1415926'); SELECT * FROM test");
 
         $this->assertSame(1, $result->getRowCount());
 
@@ -168,7 +167,7 @@ public function testMultipleQueryWithCommandResultFirst()
 
     public function testMultipleQueryWithCommandResultSecond()
     {
-        $result = $this->link->query("SELECT * FROM test; INSERT INTO test VALUES ('canon', 'jp', '{1}', true, 4.2)");
+        $result = $this->executor->query("SELECT * FROM test; INSERT INTO test VALUES ('canon', 'jp', '{1}', true, 4.2)");
 
         $data = $this->getData();
 
@@ -183,13 +182,13 @@ public function testMultipleQueryWithCommandResultSecond()
 
     public function testQueryWithUnconsumedTupleResult()
     {
-        $result = $this->link->query("SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test");
 
         $this->assertInstanceOf(Result::class, $result);
 
         unset($result); // Force destruction of result object.
 
-        $result = $this->link->query("SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test");
 
         $this->assertInstanceOf(Result::class, $result);
 
@@ -200,7 +199,7 @@ public function testQueryWithUnconsumedTupleResult()
 
     public function testQueryWithCommandResult(): void
     {
-        $result = $this->link->query("INSERT INTO test VALUES ('canon', 'jp', '{1}', true, 4.2)");
+        $result = $this->executor->query("INSERT INTO test VALUES ('canon', 'jp', '{1}', true, 4.2)");
 
         $this->assertSame(1, $result->getRowCount());
     }
@@ -208,14 +207,14 @@ public function testQueryWithCommandResult(): void
     public function testQueryWithEmptyQuery(): void
     {
         $this->expectException(QueryError::class);
-        $this->link->query('');
+        $this->executor->query('');
     }
 
     public function testQueryWithSyntaxError()
     {
         /** @var Result $result */
         try {
-            $result = $this->link->query("SELECT & FROM test");
+            $result = $this->executor->query("SELECT & FROM test");
             $this->fail(\sprintf("An instance of %s was expected to be thrown", QueryExecutionError::class));
         } catch (QueryExecutionError $exception) {
             $diagnostics = $exception->getDiagnostics();
@@ -227,7 +226,7 @@ public function testPrepare()
     {
         $query = "SELECT * FROM test WHERE domain=\$1";
 
-        $statement = $this->link->prepare($query);
+        $statement = $this->executor->prepare($query);
 
         $this->assertSame($query, $statement->getQuery());
 
@@ -242,7 +241,7 @@ public function testPrepareWithCommandResult()
     {
         $query = "INSERT INTO test (domain, tld, keys, enabled, number, json) VALUES (:domain, :tld, :keys, :enabled, :number, :json)";
 
-        $statement = $this->link->prepare($query);
+        $statement = $this->executor->prepare($query);
 
         $fields = [
             'domain' => 'canon',
@@ -259,7 +258,7 @@ public function testPrepareWithCommandResult()
 
         $this->assertSame(1, $result->getRowCount());
 
-        $result = $this->link->execute('SELECT * FROM test WHERE domain=? AND tld=?', ['canon', 'jp']);
+        $result = $this->executor->execute('SELECT * FROM test WHERE domain=? AND tld=?', ['canon', 'jp']);
 
         $this->verifyResult($result, [\array_values($fields)]);
     }
@@ -271,7 +270,7 @@ public function testPrepareWithNamedParams()
     {
         $query = "SELECT * FROM test WHERE domain=:domain AND tld=:tld";
 
-        $statement = $this->link->prepare($query);
+        $statement = $this->executor->prepare($query);
 
         $data = $this->getData()[0];
 
@@ -289,7 +288,7 @@ public function testPrepareWithUnnamedParams()
     {
         $query = "SELECT * FROM test WHERE domain=? AND tld=?";
 
-        $statement = $this->link->prepare($query);
+        $statement = $this->executor->prepare($query);
 
         $data = $this->getData()[0];
 
@@ -307,7 +306,7 @@ public function testPrepareWithNamedParamsWithDataAppearingAsNamedParam()
     {
         $query = "SELECT * FROM test WHERE domain=:domain OR domain=':domain'";
 
-        $statement = $this->link->prepare($query);
+        $statement = $this->executor->prepare($query);
 
         $data = $this->getData()[0];
 
@@ -328,7 +327,7 @@ public function testPrepareInvalidQuery()
 
         $query = "SELECT * FROM test WHERE invalid=\$1";
 
-        $statement = $this->link->prepare($query);
+        $statement = $this->executor->prepare($query);
 
         $statement->execute(['param']);
     }
@@ -340,9 +339,9 @@ public function testPrepareSameQuery()
     {
         $sql = "SELECT * FROM test WHERE domain=\$1";
 
-        $statement1 = $this->link->prepare($sql);
+        $statement1 = $this->executor->prepare($sql);
 
-        $statement2 = $this->link->prepare($sql);
+        $statement2 = $this->executor->prepare($sql);
 
         $this->assertInstanceOf(Statement::class, $statement1);
         $this->assertInstanceOf(Statement::class, $statement2);
@@ -363,8 +362,8 @@ public function testSimultaneousPrepareSameQuery()
     {
         $sql = "SELECT * FROM test WHERE domain=\$1";
 
-        $statement1 = async(fn () => $this->link->prepare($sql));
-        $statement2 = async(fn () => $this->link->prepare($sql));
+        $statement1 = async(fn () => $this->executor->prepare($sql));
+        $statement2 = async(fn () => $this->executor->prepare($sql));
 
         [$statement1, $statement2] = Future\await([$statement1, $statement2]);
 
@@ -383,9 +382,9 @@ public function testSimultaneousPrepareSameQuery()
 
     public function testPrepareSimilarQueryReturnsDifferentStatements()
     {
-        $statement1 = async(fn () => $this->link->prepare("SELECT * FROM test WHERE domain=\$1"));
+        $statement1 = async(fn () => $this->executor->prepare("SELECT * FROM test WHERE domain=\$1"));
 
-        $statement2 = async(fn () => $this->link->prepare("SELECT * FROM test WHERE domain=:domain"));
+        $statement2 = async(fn () => $this->executor->prepare("SELECT * FROM test WHERE domain=:domain"));
 
         [$statement1, $statement2] = Future\await([$statement1, $statement2]);
 
@@ -412,7 +411,7 @@ public function testPrepareSimilarQueryReturnsDifferentStatements()
 
     public function testPrepareThenExecuteWithUnconsumedTupleResult()
     {
-        $statement = $this->link->prepare("SELECT * FROM test");
+        $statement = $this->executor->prepare("SELECT * FROM test");
 
         $result = $statement->execute();
 
@@ -429,7 +428,7 @@ public function testExecute()
     {
         $data = $this->getData()[0];
 
-        $result = $this->link->execute("SELECT * FROM test WHERE domain=\$1", [$data[0]]);
+        $result = $this->executor->execute("SELECT * FROM test WHERE domain=\$1", [$data[0]]);
 
         $this->verifyResult($result, [$data]);
     }
@@ -441,7 +440,7 @@ public function testExecuteWithNamedParams()
     {
         $data = $this->getData()[0];
 
-        $result = $this->link->execute(
+        $result = $this->executor->execute(
             "SELECT * FROM test WHERE domain=:domain",
             ['domain' => $data[0]]
         );
@@ -457,7 +456,7 @@ public function testExecuteWithInvalidParams()
         $this->expectException(\Error::class);
         $this->expectExceptionMessage("Value for unnamed parameter at position 0 missing");
 
-        $this->link->execute("SELECT * FROM test WHERE domain=\$1");
+        $this->executor->execute("SELECT * FROM test WHERE domain=\$1");
     }
 
     /**
@@ -468,7 +467,7 @@ public function testExecuteWithInvalidNamedParams()
         $this->expectException(\Error::class);
         $this->expectExceptionMessage("Value for named parameter 'domain' missing");
 
-        $this->link->execute("SELECT * FROM test WHERE domain=:domain", ['tld' => 'com']);
+        $this->executor->execute("SELECT * FROM test WHERE domain=:domain", ['tld' => 'com']);
     }
 
     /**
@@ -477,7 +476,7 @@ public function testExecuteWithInvalidNamedParams()
     public function testSimultaneousQuery()
     {
         $callback = fn (int $value) => async(function () use ($value): void {
-            $result = $this->link->query("SELECT {$value} as value");
+            $result = $this->executor->query("SELECT {$value} as value");
 
             foreach ($result as $row) {
                 $this->assertEquals($value, $row['value']);
@@ -493,7 +492,7 @@ public function testSimultaneousQuery()
     public function testSimultaneousQueryWithOneFailing()
     {
         $callback = fn (string $query) => async(function () use ($query): Result {
-            $result = $this->link->query($query);
+            $result = $this->executor->query($query);
 
             $data = $this->getData();
 
@@ -527,13 +526,13 @@ public function testSimultaneousQueryAndPrepare()
     {
         $promises = [];
         $promises[] = async(function () {
-            $result = $this->link->query("SELECT * FROM test");
+            $result = $this->executor->query("SELECT * FROM test");
             $data = $this->getData();
             $this->verifyResult($result, $data);
         });
 
         $promises[] = async(function () {
-            $statement = ($this->link->prepare("SELECT * FROM test"));
+            $statement = ($this->executor->prepare("SELECT * FROM test"));
             $result = $statement->execute();
             $data = $this->getData();
             $this->verifyResult($result, $data);
@@ -545,14 +544,14 @@ public function testSimultaneousQueryAndPrepare()
     public function testSimultaneousPrepareAndExecute()
     {
         $promises[] = async(function () {
-            $statement = $this->link->prepare("SELECT * FROM test");
+            $statement = $this->executor->prepare("SELECT * FROM test");
             $result = $statement->execute();
             $data = $this->getData();
             $this->verifyResult($result, $data);
         });
 
         $promises[] = async(function () {
-            $result = $this->link->execute("SELECT * FROM test");
+            $result = $this->executor->execute("SELECT * FROM test");
             $data = $this->getData();
             $this->verifyResult($result, $data);
         });
@@ -562,9 +561,7 @@ public function testSimultaneousPrepareAndExecute()
 
     public function testTransaction()
     {
-        $isolation = TransactionIsolationLevel::Committed;
-
-        $transaction = $this->link->beginTransaction($isolation);
+        $transaction = $this->executor->beginTransaction();
 
         $this->assertInstanceOf(PostgresTransaction::class, $transaction);
 
@@ -572,20 +569,19 @@ public function testTransaction()
 
         $this->assertFalse($transaction->isClosed());
         $this->assertTrue($transaction->isActive());
-        $this->assertSame($isolation, $transaction->getIsolationLevel());
 
-        $transaction->createSavepoint('test');
+        $nested = $transaction->beginTransaction();
 
-        $statement = $transaction->prepare("SELECT * FROM test WHERE domain=:domain");
+        $statement = $nested->prepare("SELECT * FROM test WHERE domain=:domain");
         $result = $statement->execute(['domain' => $data[0]]);
 
         unset($result); // Force destruction of result object.
 
-        $result = $transaction->execute("SELECT * FROM test WHERE domain=\$1 FOR UPDATE", [$data[0]]);
+        $result = $nested->execute("SELECT * FROM test WHERE domain=\$1 FOR UPDATE", [$data[0]]);
 
         unset($result); // Force destruction of result object.
 
-        $transaction->rollbackTo('test');
+        $nested->rollback();
 
         $transaction->commit();
 
@@ -630,13 +626,13 @@ public function testStatementInsertByteA(
         array $params,
         array $expected
     ): void {
-        $statement = $this->link->prepare($insertSql);
+        $statement = $this->executor->prepare($insertSql);
 
         $result = $statement->execute($params);
 
         $this->assertSame(1, $result->getRowCount());
 
-        $result = $this->link->execute($selectSql, $params);
+        $result = $this->executor->execute($selectSql, $params);
         $this->assertSame($expected, $result->fetchRow());
     }
 
@@ -649,11 +645,11 @@ public function testExecuteInsertByteA(
         array $params,
         array $expected
     ): void {
-        $result = $this->link->execute($insertSql, $params);
+        $result = $this->executor->execute($insertSql, $params);
 
         $this->assertSame(1, $result->getRowCount());
 
-        $result = $this->link->execute($selectSql, $params);
+        $result = $this->executor->execute($selectSql, $params);
         $this->assertSame($expected, $result->fetchRow());
     }
 }
diff --git a/test/AbstractQuoteTest.php b/test/AbstractQuoteTest.php
index eac1a9d..07a0566 100644
--- a/test/AbstractQuoteTest.php
+++ b/test/AbstractQuoteTest.php
@@ -2,12 +2,12 @@
 
 namespace Amp\Postgres\Test;
 
+use Amp\Postgres\Internal\PostgresHandleConnection;
 use Amp\Postgres\PostgresConfig;
-use Amp\Postgres\PostgresConnection;
 
 abstract class AbstractQuoteTest extends AbstractConnectTest
 {
-    private PostgresConnection $connection;
+    private PostgresHandleConnection $connection;
 
     public function setUp(): void
     {
diff --git a/test/FunctionsTest.php b/test/FunctionsTest.php
index a612ff1..aef26e5 100644
--- a/test/FunctionsTest.php
+++ b/test/FunctionsTest.php
@@ -5,8 +5,8 @@
 use Amp\CancelledException;
 use Amp\DeferredCancellation;
 use Amp\PHPUnit\AsyncTestCase;
+use Amp\Postgres\Internal\PostgresHandleConnection;
 use Amp\Postgres\PostgresConfig;
-use Amp\Postgres\PostgresConnection;
 use function Amp\Postgres\connect;
 
 class FunctionsTest extends AsyncTestCase
@@ -23,7 +23,7 @@ public function setUp(): void
     public function testConnect()
     {
         $connection = connect(PostgresConfig::fromString('host=localhost user=postgres password=postgres'));
-        $this->assertInstanceOf(PostgresConnection::class, $connection);
+        $this->assertInstanceOf(PostgresHandleConnection::class, $connection);
     }
 
     /**
diff --git a/test/PgSqlConnectionTest.php b/test/PgSqlConnectionTest.php
index 61d1657..5287866 100644
--- a/test/PgSqlConnectionTest.php
+++ b/test/PgSqlConnectionTest.php
@@ -4,7 +4,8 @@
 
 use Amp\Postgres\ByteA;
 use Amp\Postgres\PgSqlConnection;
-use Amp\Postgres\PostgresLink;
+use Amp\Postgres\PostgresConfig;
+use Amp\Postgres\PostgresExecutor;
 use Revolt\EventLoop;
 use function Amp\Postgres\Internal\cast;
 
@@ -15,7 +16,7 @@ class PgSqlConnectionTest extends AbstractConnectionTest
 {
     protected ?\PgSql\Connection $handle = null;
 
-    public function createLink(string $connectionString): PostgresLink
+    public function createExecutor(string $connectionString): PostgresExecutor
     {
         if (EventLoop::getDriver()->getHandle() instanceof \EvLoop) {
             $this->markTestSkipped("ext-pgsql is not compatible with pecl-ev");
@@ -39,7 +40,13 @@ public function createLink(string $connectionString): PostgresLink
             }
         }
 
-        return $this->newConnection(PgSqlConnection::class, $this->handle, $socket, 'mock-connection');
+        return $this->newConnection(
+            PgSqlConnection::class,
+            $this->handle,
+            $socket,
+            'mock-connection',
+            PostgresConfig::fromString($connectionString),
+        );
     }
 
     private function cast(mixed $param): mixed
diff --git a/test/PgSqlNestedTransactionTest.php b/test/PgSqlNestedTransactionTest.php
index 36fda6f..1043445 100644
--- a/test/PgSqlNestedTransactionTest.php
+++ b/test/PgSqlNestedTransactionTest.php
@@ -4,8 +4,7 @@
 
 use Amp\Postgres\PgSqlConnection;
 use Amp\Postgres\PostgresConfig;
-use Amp\Postgres\PostgresLink;
-use Amp\Postgres\PostgresNestableTransaction;
+use Amp\Postgres\PostgresExecutor;
 use Amp\Postgres\PostgresTransaction;
 
 /**
@@ -16,7 +15,7 @@ class PgSqlNestedTransactionTest extends AbstractLinkTest
     protected PgSqlConnection $connection;
     protected PostgresTransaction $transaction;
 
-    public function createLink(string $connectionString): PostgresLink
+    public function createExecutor(string $connectionString): PostgresExecutor
     {
         $connectionConfig = PostgresConfig::fromString($connectionString);
         $connection = PgSqlConnection::connect($connectionConfig);
@@ -32,7 +31,7 @@ public function createLink(string $connectionString): PostgresLink
         $this->connection = $connection;
         $this->transaction = $connection->beginTransaction();
 
-        return new PostgresNestableTransaction($this->transaction);
+        return $this->transaction->beginTransaction();
     }
 
     public function tearDown(): void
diff --git a/test/PgSqlPoolTest.php b/test/PgSqlPoolTest.php
index 91843a5..7004d49 100644
--- a/test/PgSqlPoolTest.php
+++ b/test/PgSqlPoolTest.php
@@ -6,7 +6,7 @@
 use Amp\Postgres\PgSqlConnection;
 use Amp\Postgres\PostgresConfig;
 use Amp\Postgres\PostgresConnectionPool;
-use Amp\Postgres\PostgresLink;
+use Amp\Postgres\PostgresExecutor;
 use Amp\Sql\Common\ConnectionPool;
 use Amp\Sql\SqlConnector;
 use Revolt\EventLoop;
@@ -22,7 +22,7 @@ class PgSqlPoolTest extends AbstractConnectionTest
     /** @var \PgSql\Connection[] PostgreSQL connection resources. */
     protected array $handles = [];
 
-    public function createLink(string $connectionString): PostgresLink
+    public function createExecutor(string $connectionString): PostgresExecutor
     {
         if (EventLoop::getDriver()->getHandle() instanceof \EvLoop) {
             $this->markTestSkipped("ext-pgsql is not compatible with pecl-ev");
@@ -32,16 +32,24 @@ public function createLink(string $connectionString): PostgresLink
             $this->handles[] = \pg_connect($connectionString, \PGSQL_CONNECT_FORCE_NEW);
         }
 
+        $config = PostgresConfig::fromString($connectionString);
+
         $connector = $this->createMock(SqlConnector::class);
         $connector->method('connect')
-            ->willReturnCallback(function (): PgSqlConnection {
+            ->willReturnCallback(function () use ($config): PgSqlConnection {
                 static $count = 0;
                 if (!isset($this->handles[$count])) {
                     $this->fail("createConnection called too many times");
                 }
                 $handle = $this->handles[$count];
                 ++$count;
-                return $this->newConnection(PgSqlConnection::class, $handle, \pg_socket($handle), 'mock-connection');
+                return $this->newConnection(
+                    PgSqlConnection::class,
+                    $handle,
+                    \pg_socket($handle),
+                    'mock-connection',
+                    $config,
+                );
             });
 
         $pool = new PostgresConnectionPool(new PostgresConfig('localhost'), \count($this->handles), ConnectionPool::DEFAULT_IDLE_TIMEOUT, true, $connector);
diff --git a/test/PqConnectionTest.php b/test/PqConnectionTest.php
index 10a26e4..6eee7d7 100644
--- a/test/PqConnectionTest.php
+++ b/test/PqConnectionTest.php
@@ -5,7 +5,8 @@
 use Amp\Postgres\ByteA;
 use Amp\Postgres\Internal\PqBufferedResultSet;
 use Amp\Postgres\Internal\PqUnbufferedResultSet;
-use Amp\Postgres\PostgresLink;
+use Amp\Postgres\PostgresConfig;
+use Amp\Postgres\PostgresExecutor;
 use Amp\Postgres\PqConnection;
 use function Amp\Postgres\Internal\cast;
 
@@ -17,7 +18,7 @@ class PqConnectionTest extends AbstractConnectionTest
     /** @var \pg\Connection|null PostgreSQL connection resource. */
     protected ?\pq\Connection $handle;
 
-    public function createLink(string $connectionString): PostgresLink
+    public function createExecutor(string $connectionString): PostgresExecutor
     {
         $this->handle = new \pq\Connection($connectionString);
         $this->handle->nonblocking = true;
@@ -39,7 +40,7 @@ public function createLink(string $connectionString): PostgresLink
             }
         }
 
-        return $this->newConnection(PqConnection::class, $this->handle);
+        return $this->newConnection(PqConnection::class, $this->handle, PostgresConfig::fromString($connectionString));
     }
 
     private function cast(mixed $param): mixed
@@ -60,12 +61,12 @@ public function tearDown(): void
 
     public function testBufferedResults(): void
     {
-        \assert($this->link instanceof PqConnection);
-        $this->link->shouldBufferResults();
+        \assert($this->executor instanceof PqConnection);
+        $this->executor->shouldBufferResults();
 
-        $this->assertTrue($this->link->isBufferingResults());
+        $this->assertTrue($this->executor->isBufferingResults());
 
-        $result = $this->link->query("SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test");
         \assert($result instanceof PqBufferedResultSet);
 
         $data = $this->getData();
@@ -77,12 +78,12 @@ public function testBufferedResults(): void
      */
     public function testUnbufferedResults(): void
     {
-        \assert($this->link instanceof PqConnection);
-        $this->link->shouldNotBufferResults();
+        \assert($this->executor instanceof PqConnection);
+        $this->executor->shouldNotBufferResults();
 
-        $this->assertFalse($this->link->isBufferingResults());
+        $this->assertFalse($this->executor->isBufferingResults());
 
-        $result = $this->link->query("SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test");
         \assert($result instanceof PqUnbufferedResultSet);
 
         $data = $this->getData();
@@ -91,7 +92,7 @@ public function testUnbufferedResults(): void
 
     public function testNextResultBeforeConsumption()
     {
-        $result = $this->link->query("SELECT * FROM test; SELECT * FROM test;");
+        $result = $this->executor->query("SELECT * FROM test; SELECT * FROM test;");
 
         $result = $result->getNextResult();
 
@@ -100,11 +101,11 @@ public function testNextResultBeforeConsumption()
 
     public function testUnconsumedMultiResult()
     {
-        $result = $this->link->query("SELECT * FROM test; SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test; SELECT * FROM test");
 
         unset($result);
 
-        $result = $this->link->query("SELECT * FROM test; SELECT * FROM test");
+        $result = $this->executor->query("SELECT * FROM test; SELECT * FROM test");
 
         $this->verifyResult($result, $this->getData());
 
diff --git a/test/PqNestedTransactionTest.php b/test/PqNestedTransactionTest.php
index edc9178..30b2b3b 100644
--- a/test/PqNestedTransactionTest.php
+++ b/test/PqNestedTransactionTest.php
@@ -3,8 +3,7 @@
 namespace Amp\Postgres\Test;
 
 use Amp\Postgres\PostgresConfig;
-use Amp\Postgres\PostgresLink;
-use Amp\Postgres\PostgresNestableTransaction;
+use Amp\Postgres\PostgresExecutor;
 use Amp\Postgres\PostgresTransaction;
 use Amp\Postgres\PqConnection;
 
@@ -15,7 +14,7 @@ class PqNestedTransactionTest extends AbstractLinkTest
 {
     protected PostgresTransaction $transaction;
 
-    public function createLink(string $connectionString): PostgresLink
+    public function createExecutor(string $connectionString): PostgresExecutor
     {
         $connectionConfig = PostgresConfig::fromString($connectionString);
         $connection = PqConnection::connect($connectionConfig);
@@ -30,7 +29,7 @@ public function createLink(string $connectionString): PostgresLink
 
         $this->transaction = $connection->beginTransaction();
 
-        return new PostgresNestableTransaction($this->transaction);
+        return $this->transaction->beginTransaction();
     }
 
     public function tearDown(): void
diff --git a/test/PqPoolTest.php b/test/PqPoolTest.php
index e16d307..f0d0a35 100644
--- a/test/PqPoolTest.php
+++ b/test/PqPoolTest.php
@@ -5,7 +5,7 @@
 use Amp\Postgres\ByteA;
 use Amp\Postgres\PostgresConfig;
 use Amp\Postgres\PostgresConnectionPool;
-use Amp\Postgres\PostgresLink;
+use Amp\Postgres\PostgresExecutor;
 use Amp\Postgres\PqConnection;
 use Amp\Sql\Common\ConnectionPool;
 use Amp\Sql\SqlConnector;
@@ -21,7 +21,7 @@ class PqPoolTest extends AbstractConnectionTest
     /** @var \pq\Connection[] */
     protected array $handles = [];
 
-    public function createLink(string $connectionString): PostgresLink
+    public function createExecutor(string $connectionString): PostgresExecutor
     {
         for ($i = 0; $i < self::POOL_SIZE; ++$i) {
             $this->handles[] = $handle = new \pq\Connection($connectionString);
@@ -29,16 +29,18 @@ public function createLink(string $connectionString): PostgresLink
             $handle->unbuffered = true;
         }
 
+        $config = PostgresConfig::fromString($connectionString);
+
         $connector = $this->createMock(SqlConnector::class);
         $connector->method('connect')
-            ->willReturnCallback(function (): PqConnection {
+            ->willReturnCallback(function () use ($config): PqConnection {
                 static $count = 0;
                 if (!isset($this->handles[$count])) {
                     $this->fail("createConnection called too many times");
                 }
                 $handle = $this->handles[$count];
                 ++$count;
-                return $this->newConnection(PqConnection::class, $handle);
+                return $this->newConnection(PqConnection::class, $handle, $config);
             });
 
         $pool = new PostgresConnectionPool(new PostgresConfig('localhost'), \count($this->handles), ConnectionPool::DEFAULT_IDLE_TIMEOUT, true, $connector);