diff --git a/composer.json b/composer.json
index 1a937e3..9834468 100644
--- a/composer.json
+++ b/composer.json
@@ -2,9 +2,10 @@
     "name": "amphp/mysql-dbal",
     "type": "library",
     "require": {
+        "php": ">=8.1",
         "amphp/amp": "^3",
         "amphp/mysql": "^3",
-        "doctrine/dbal": "^3"
+        "doctrine/dbal": "^4"
     },
     "license": "MIT",
     "authors": [
diff --git a/src/MysqlConnection.php b/src/MysqlConnection.php
index 48d38ed..bac33e4 100644
--- a/src/MysqlConnection.php
+++ b/src/MysqlConnection.php
@@ -2,32 +2,41 @@
 
 namespace Amp\Mysql\DBAL;
 
-use Amp\Mysql\Connection as SqlConnection;
-use Amp\Mysql\Result as SqlResult;
+use Amp\Mysql\MysqlConnection as AmphpMysqlConnection;
+use Amp\Mysql\MysqlResult as AmphpMysqlResult;
+use Closure;
+use Doctrine\DBAL\Driver\Connection;
 use Doctrine\DBAL\Driver\Result;
-use Doctrine\DBAL\Driver\ServerInfoAwareConnection;
 use Doctrine\DBAL\Driver\Statement;
 use Doctrine\DBAL\ParameterType;
-use function Amp\await;
-use function Amp\Pipeline\discard;
+use Error;
+use Throwable;
+use function Amp\Future\await;
 
-class MysqlConnection implements ServerInfoAwareConnection
+class MysqlConnection implements Connection
 {
-    private SqlConnection $connection;
-    private \Closure $resultListener;
-    private mixed $lastInsertId;
+    private AmphpMysqlConnection $connection;
 
-    public function __construct(SqlConnection $connection)
+    private Closure $resultListener;
+
+    private int|string|null $lastInsertId = null;
+
+    public function __construct(AmphpMysqlConnection $connection)
     {
         $this->connection = $connection;
-        $this->resultListener = fn(SqlResult $result) => $this->lastInsertId = $result->getLastInsertId();
+        $this->resultListener = fn(AmphpMysqlResult $result) => $this->lastInsertId = $result->getLastInsertId();
+    }
+
+    public function getNativeConnection(): AmphpMysqlConnection
+    {
+        return $this->connection;
     }
 
     public function prepare(string $sql): Statement
     {
         try {
             return new MysqlStatement($this->connection->prepare($sql), $this->resultListener);
-        } catch (\Throwable $e) {
+        } catch (Throwable $e) {
             throw MysqlException::new($e);
         }
     }
@@ -39,14 +48,14 @@ public function query(string $sql): Result
             ($this->resultListener)($result);
 
             return new MysqlResult($result);
-        } catch (\Throwable $e) {
+        } catch (Throwable $e) {
             throw MysqlException::new($e);
         }
     }
 
-    public function quote($value, $type = ParameterType::STRING)
+    public function quote($value, $type = ParameterType::STRING): string
     {
-        throw new \Error("Not implemented, use prepared statements");
+        throw new Error('Not implemented, use prepared statements');
     }
 
     public function exec(string $sql): int
@@ -56,51 +65,45 @@ public function exec(string $sql): int
             ($this->resultListener)($result);
 
             return $result->getRowCount();
-        } catch (\Throwable $e) {
+        } catch (Throwable $e) {
             throw MysqlException::new($e);
         }
     }
 
-    public function lastInsertId($name = null)
+    public function lastInsertId($name = null): int|string
     {
         return $this->lastInsertId;
     }
 
-    public function beginTransaction(): bool
+    public function beginTransaction(): void
     {
         try {
-            await(discard($this->connection->query("START TRANSACTION")));
-
-            return true;
-        } catch (\Throwable $e) {
+            await($this->connection->query('START TRANSACTION'));
+        } catch (Throwable $e) {
             throw MysqlException::new($e);
         }
     }
 
-    public function commit(): bool
+    public function commit(): void
     {
         try {
-            await(discard($this->connection->query("COMMIT")));
-
-            return true;
-        } catch (\Throwable $e) {
+            await($this->connection->query('COMMIT'));
+        } catch (Throwable $e) {
             throw MysqlException::new($e);
         }
     }
 
-    public function rollBack(): bool
+    public function rollBack(): void
     {
         try {
-            await(discard($this->connection->query("ROLLBACK")));
-
-            return true;
-        } catch (\Throwable $e) {
+            await($this->connection->query('ROLLBACK'));
+        } catch (Throwable $e) {
             throw MysqlException::new($e);
         }
     }
 
     public function getServerVersion(): string
     {
-        return $this->query("SELECT @@version")->fetchOne();
+        return $this->query('SELECT @@version')->fetchOne();
     }
-}
\ No newline at end of file
+}
diff --git a/src/MysqlDriver.php b/src/MysqlDriver.php
index 1ad2901..d48846c 100644
--- a/src/MysqlDriver.php
+++ b/src/MysqlDriver.php
@@ -2,31 +2,36 @@
 
 namespace Amp\Mysql\DBAL;
 
-use Amp\Mysql\CancellableConnector;
-use Amp\Mysql\ConnectionConfig;
-use Amp\Socket\StaticConnector;
+use Amp\Mysql\MysqlConfig;
+use Amp\Mysql\SocketMysqlConnector;
 use Doctrine\DBAL\Driver;
 use Doctrine\DBAL\Driver\Connection;
-use function Amp\Socket\connector;
+use Throwable;
+use function Amp\Mysql\mysqlConnector;
 
 final class MysqlDriver extends Driver\AbstractMySQLDriver
 {
     public function connect(array $params): Connection
     {
-        $config = new ConnectionConfig($params['host'] ?? 'localhost',
-            $params['port'] ?? ConnectionConfig::DEFAULT_PORT,
-            $params['user'] ?? '', $params['password'] ?? '', $params['dbname'] ?? null, null,
-            $params['charset'] ?? ConnectionConfig::DEFAULT_CHARSET);
+        $config = new MysqlConfig(
+            $params['host'] ?? 'localhost',
+            $params['port'] ?? MysqlConfig::DEFAULT_PORT,
+            $params['user'] ?? '',
+            $params['password'] ?? '',
+            $params['dbname'] ?? null,
+            null,
+            $params['charset'] ?? MysqlConfig::DEFAULT_CHARSET
+        );
 
-        $connector = connector();
+        $connector = mysqlConnector();
         if (isset($params['unix_socket'])) {
-            $connector = new StaticConnector('unix:' . $params['unix_socket'], $connector);
+            $connector = new SocketMysqlConnector();
         }
 
         try {
-            return new MysqlConnection((new CancellableConnector($connector))->connect($config));
-        } catch (\Throwable $e) {
+            return new MysqlConnection($connector->connect($config));
+        } catch (Throwable $e) {
             throw MysqlException::new($e);
         }
     }
-}
\ No newline at end of file
+}
diff --git a/src/MysqlException.php b/src/MysqlException.php
index 7259ef5..9cd7f70 100644
--- a/src/MysqlException.php
+++ b/src/MysqlException.php
@@ -3,11 +3,12 @@
 namespace Amp\Mysql\DBAL;
 
 use Doctrine\DBAL\Driver\AbstractException;
+use Throwable;
 
 final class MysqlException extends AbstractException
 {
-    public static function new(\Throwable $exception): self
+    public static function new(Throwable $exception): self
     {
         return new self($exception->getMessage(), null, $exception->getCode(), $exception);
     }
-}
\ No newline at end of file
+}
diff --git a/src/MysqlResult.php b/src/MysqlResult.php
index 320c85b..3473491 100644
--- a/src/MysqlResult.php
+++ b/src/MysqlResult.php
@@ -2,9 +2,11 @@
 
 namespace Amp\Mysql\DBAL;
 
-use Amp\Mysql\Result as SqlResult;
+use Amp\Mysql\MysqlResult as SqlResult;
 use Doctrine\DBAL\Driver\FetchUtils;
 use Doctrine\DBAL\Driver\Result;
+use function array_values;
+use function count;
 
 class MysqlResult implements Result
 {
@@ -22,16 +24,16 @@ public function fetchNumeric(): array|false
             return false;
         }
 
-        return \array_values($row);
+        return array_values($row);
     }
 
     public function fetchAssociative(): array|false
     {
         /** @noinspection ProperNullCoalescingOperatorUsageInspection */
-        return $this->result->continue() ?? false;
+        return $this->result->fetchRow() ?? false;
     }
 
-    public function fetchOne()
+    public function fetchOne(): mixed
     {
         return FetchUtils::fetchOne($this);
     }
@@ -58,11 +60,10 @@ public function rowCount(): int
 
     public function columnCount(): int
     {
-        return \count($this->result->getFields());
+        return count($this->result->getColumnDefinitions());
     }
 
     public function free(): void
     {
-        $this->result->dispose();
     }
-}
\ No newline at end of file
+}
diff --git a/src/MysqlStatement.php b/src/MysqlStatement.php
index b0d9d1e..ec8aef5 100644
--- a/src/MysqlStatement.php
+++ b/src/MysqlStatement.php
@@ -2,63 +2,38 @@
 
 namespace Amp\Mysql\DBAL;
 
-use Amp\Mysql\Statement as SqlStatement;
+use Amp\Mysql\MysqlStatement as SqlStatement;
+use Closure;
 use Doctrine\DBAL\Driver\Exception;
 use Doctrine\DBAL\Driver\Result;
 use Doctrine\DBAL\Driver\Statement;
 use Doctrine\DBAL\ParameterType;
+use Throwable;
+use function is_int;
 
 class MysqlStatement implements Statement
 {
-    private const PARAM_TYPES = [
-        ParameterType::NULL => null,
-        ParameterType::INTEGER => null,
-        ParameterType::STRING => null,
-        ParameterType::ASCII => null,
-        ParameterType::BINARY => null,
-        ParameterType::LARGE_OBJECT => null,
-        ParameterType::BOOLEAN => null,
-    ];
-
     private SqlStatement $statement;
-    private \Closure $resultListener;
+
+    private Closure $resultListener;
 
     private array $values = [];
+
     private array $types = [];
 
     public function __construct(SqlStatement $statement, callable $resultListener)
     {
         $this->statement = $statement;
-        $this->resultListener = $resultListener instanceof \Closure
+        $this->resultListener = $resultListener instanceof Closure
             ? $resultListener
-            : \Closure::fromCallable($resultListener);
+            : $resultListener(...);
     }
 
-    public function bindValue($param, $value, $type = ParameterType::STRING): bool
+    public function bindValue($param, $value, ParameterType $type = ParameterType::STRING): void
     {
-        if (!isset(self::PARAM_TYPES[$type])) {
-            throw Exception\UnknownParameterType::new($type);
-        }
-
-        $key = \is_int($param) ? $param - 1 : $param;
+        $key = is_int($param) ? $param - 1 : $param;
 
         $this->values[$key] = $this->convertValue($value, $type);
-
-        return true;
-    }
-
-    public function bindParam($param, &$variable, $type = ParameterType::STRING, $length = null): bool
-    {
-        if (!isset(self::PARAM_TYPES[$type])) {
-            throw Exception\UnknownParameterType::new($type);
-        }
-
-        $key = \is_int($param) ? $param - 1 : $param;
-
-        $this->values[$key] = &$variable;
-        $this->types[$key] = $type;
-
-        return true;
     }
 
     public function execute($params = null): Result
@@ -81,19 +56,18 @@ public function execute($params = null): Result
             ($this->resultListener)($result);
 
             return new MysqlResult($result);
-        } catch (\Throwable $e) {
+        } catch (Throwable $e) {
             throw MysqlException::new($e);
         }
     }
 
-    private function convertValue($value, int $type): null|bool|int|string
+    private function convertValue($value, ParameterType $type): null|bool|int|string
     {
         return match ($type) {
-            ParameterType::NULL => null,
+            ParameterType::STRING, ParameterType::ASCII, ParameterType::LARGE_OBJECT, ParameterType::BINARY => (string) $value,
             ParameterType::INTEGER => (int) $value,
-            ParameterType::ASCII, ParameterType::LARGE_OBJECT, ParameterType::BINARY, ParameterType::STRING => (string) $value,
             ParameterType::BOOLEAN => (bool) $value,
-            default => throw Exception\UnknownParameterType::new($type),
+            ParameterType::NULL => null,
         };
     }
-}
\ No newline at end of file
+}