diff --git a/src/ORM/Connect/Database.php b/src/ORM/Connect/Database.php
index 729770895..fa32db16d 100644
--- a/src/ORM/Connect/Database.php
+++ b/src/ORM/Connect/Database.php
@@ -583,6 +583,17 @@ abstract class Database
      */
     abstract public function supportsTransactions();
 
+    /**
+     * Does this database support savepoints in transactions
+     * By default it is assumed that they don't unless they are explicitly enabled.
+     *
+     * @return boolean Flag indicating support for savepoints in transactions
+     */
+    public function supportsSavepoints()
+    {
+        return false;
+    }
+
     /**
      * Invoke $callback within a transaction
      *
diff --git a/src/ORM/Connect/MySQLDatabase.php b/src/ORM/Connect/MySQLDatabase.php
index 74f155e30..c19004885 100644
--- a/src/ORM/Connect/MySQLDatabase.php
+++ b/src/ORM/Connect/MySQLDatabase.php
@@ -21,7 +21,7 @@ use Exception;
  * You are advised to backup your tables if changing settings on an existing database
  * `connection_charset` and `charset` should be equal, similarly so should `connection_collation` and `collation`
  */
-class MySQLDatabase extends Database
+class MySQLDatabase extends Database implements TransactionManager
 {
     use Configurable;
 
@@ -49,6 +49,13 @@ class MySQLDatabase extends Database
      */
     private static $charset = 'utf8';
 
+    /**
+     * Cache for getTransactionManager()
+     *
+     * @var TransactionManager
+     */
+    private $transactionManager = null;
+
     /**
      * Default collation
      *
@@ -57,11 +64,6 @@ class MySQLDatabase extends Database
      */
     private static $collation = 'utf8_general_ci';
 
-    /**
-     * @var bool
-     */
-    protected $transactionNesting = 0;
-
     public function connect($parameters)
     {
         // Ensure that driver is available (required by PDO)
@@ -298,73 +300,64 @@ class MySQLDatabase extends Database
         return $list;
     }
 
+
+    /**
+     * Returns the TransactionManager to handle transactions for this database.
+     *
+     * @return TransactionManager
+     */
+    protected function getTransactionManager()
+    {
+        if (!$this->transactionManager) {
+            // PDOConnector providers this
+            if ($this->connector instanceof TransactionManager) {
+                $this->transactionManager = new NestedTransactionManager($this->connector);
+            // Direct database access does not
+            } else {
+                $this->transactionManager = new NestedTransactionManager(new MySQLTransactionManager($this));
+            }
+        }
+        return $this->transactionManager;
+    }
     public function supportsTransactions()
     {
         return true;
     }
+    public function supportsSavepoints()
+    {
+        return $this->getTransactionManager()->supportsSavepoints();
+    }
 
     public function transactionStart($transactionMode = false, $sessionCharacteristics = false)
     {
-        if ($this->transactionNesting > 0) {
-            $this->transactionSavepoint('NESTEDTRANSACTION' . $this->transactionNesting);
-        } else {
-            // This sets the isolation level for the NEXT transaction, not the current one.
-            if ($transactionMode) {
-                $this->query('SET TRANSACTION ' . $transactionMode);
-            }
-
-            $this->query('START TRANSACTION');
-
-            if ($sessionCharacteristics) {
-                $this->query('SET SESSION TRANSACTION ' . $sessionCharacteristics);
-            }
-        }
-        ++$this->transactionNesting;
+        $this->getTransactionManager()->transactionStart($transactionMode, $sessionCharacteristics);
     }
 
     public function transactionSavepoint($savepoint)
     {
-        $this->query("SAVEPOINT $savepoint");
+        $this->getTransactionManager()->transactionSavepoint($savepoint);
     }
 
     public function transactionRollback($savepoint = false)
     {
-        // Named transaction
-        if ($savepoint) {
-            $this->query('ROLLBACK TO ' . $savepoint);
-            return true;
-        }
-
-        // Fail if transaction isn't available
-        if (!$this->transactionNesting) {
-            return false;
-        }
-        --$this->transactionNesting;
-        if ($this->transactionNesting > 0) {
-            $this->transactionRollback('NESTEDTRANSACTION' . $this->transactionNesting);
-        } else {
-            $this->query('ROLLBACK');
-        }
-        return true;
+        return $this->getTransactionManager()->transactionRollback($savepoint);
     }
 
     public function transactionDepth()
     {
-        return $this->transactionNesting;
+        return $this->getTransactionManager()->transactionDepth();
     }
 
     public function transactionEnd($chain = false)
     {
-        // Fail if transaction isn't available
-        if (!$this->transactionNesting) {
-            return false;
+        $result = $this->getTransactionManager()->transactionEnd();
+
+        if ($chain) {
+            Deprecation::notice('4.4', '$chain argument is deprecated');
+            return $this->getTransactionManager()->transactionStart();
         }
-        --$this->transactionNesting;
-        if ($this->transactionNesting <= 0) {
-            $this->transactionNesting = 0;
-            $this->query('COMMIT AND ' . ($chain ? '' : 'NO ') . 'CHAIN');
-        }
-        return true;
+
+        return $result;
     }
 
     /**
@@ -372,6 +365,12 @@ class MySQLDatabase extends Database
      */
     protected function resetTransactionNesting()
     {
+        // Check whether to use a connector's built-in transaction methods
+        if ($this->connector instanceof TransactionalDBConnector) {
+            if ($this->transactionNesting > 0) {
+                $this->connector->transactionRollback();
+            }
+        }
         $this->transactionNesting = 0;
     }
 
diff --git a/src/ORM/Connect/MySQLTransactionManager.php b/src/ORM/Connect/MySQLTransactionManager.php
new file mode 100644
index 000000000..452ce334a
--- /dev/null
+++ b/src/ORM/Connect/MySQLTransactionManager.php
@@ -0,0 +1,100 @@
+<?php
+
+namespace SilverStripe\ORM\Connect;
+
+use SilverStripe\Dev\Deprecation;
+
+/**
+ * TransactionManager that executes MySQL-compatible transaction control queries
+ */
+class MySQLTransactionManager implements TransactionManager
+{
+    protected $dbConn;
+
+    protected $inTransaction = false;
+
+    public function __construct(Database $dbConn)
+    {
+        $this->dbConn = $dbConn;
+    }
+
+    public function transactionStart($transactionMode = false, $sessionCharacteristics = false)
+    {
+        if ($transactionMode || $sessionCharacteristics) {
+            Deprecation::notice(
+                '4.4',
+                '$transactionMode and $sessionCharacteristics are deprecated and will be removed in SS5'
+            );
+        }
+
+        if ($this->inTransaction) {
+            throw new DatabaseException(
+                "Already in transaction, can't start another. Consider decorating with NestedTransactionManager."
+            );
+        }
+
+        // This sets the isolation level for the NEXT transaction, not the current one.
+        if ($transactionMode) {
+            $this->dbConn->query('SET TRANSACTION ' . $transactionMode);
+        }
+
+        $this->dbConn->query('START TRANSACTION');
+
+        if ($sessionCharacteristics) {
+            $this->dbConn->query('SET SESSION TRANSACTION ' . $sessionCharacteristics);
+        }
+
+        $this->inTransaction = true;
+        return true;
+    }
+
+    public function transactionEnd($chain = false)
+    {
+        if (!$this->inTransaction) {
+            throw new DatabaseException("Not in transaction, can't end.");
+        }
+
+        if ($chain) {
+            user_error(
+                "transactionEnd() chain argument no longer implemented. Use NestedTransactionManager",
+                E_USER_WARNING
+            );
+        }
+
+        $this->dbConn->query('COMMIT');
+
+        $this->inTransaction = false;
+        return true;
+    }
+
+    public function transactionRollback($savepoint = null)
+    {
+        if (!$this->inTransaction) {
+            throw new DatabaseException("Not in transaction, can't roll back.");
+        }
+
+        if ($savepoint) {
+            $this->dbConn->query("ROLLBACK TO SAVEPOINT $savepoint");
+        } else {
+            $this->dbConn->query('ROLLBACK');
+            $this->inTransaction = false;
+        }
+
+        return true;
+    }
+
+    public function transactionSavepoint($savepoint)
+    {
+        $this->dbConn->query("SAVEPOINT $savepoint");
+    }
+
+    public function transactionDepth()
+    {
+        return (int)$this->inTransaction;
+    }
+
+    public function supportsSavepoints()
+    {
+        return true;
+    }
+}
diff --git a/src/ORM/Connect/NestedTransactionManager.php b/src/ORM/Connect/NestedTransactionManager.php
new file mode 100644
index 000000000..46e8090a3
--- /dev/null
+++ b/src/ORM/Connect/NestedTransactionManager.php
@@ -0,0 +1,127 @@
+<?php
+
+namespace SilverStripe\ORM\Connect;
+
+/**
+ * TransactionManager decorator that adds virtual nesting support.
+ * Because this is managed in PHP and not the database, it has the following limitations:
+ *   - Committing a nested transaction won't change anything until the parent transaction is committed
+ *   - Rolling back a nested transaction means that the parent transaction must be rolled backed
+ *
+ * DBAL describes this behaviour nicely in their docs: https://www.doctrine-project.org/projects/doctrine-dbal/en/2.8/reference/transactions.html#transaction-nesting
+ */
+
+class NestedTransactionManager implements TransactionManager
+{
+
+    /**
+     * @var int
+     */
+    protected $transactionNesting = 0;
+
+    /**
+     * @var TransactionManager
+     */
+    protected $child;
+
+    /**
+     * Set to true if all transactions must roll back to the parent
+     * @var boolean
+     */
+    protected $mustRollback = false;
+
+    /**
+     * Create a NestedTransactionManager
+     * @param TransactionManager $child The transaction manager that will handle the topmost transaction
+     */
+    public function __construct(TransactionManager $child)
+    {
+        $this->child = $child;
+    }
+
+    /**
+     * Start a transaction
+     * @throws DatabaseException on failure
+     * @return bool True on success
+     */
+    public function transactionStart($transactionMode = false, $sessionCharacteristics = false)
+    {
+        if ($this->transactionNesting <= 0) {
+            $this->transactionNesting = 1;
+            $this->child->transactionStart($transactionMode, $sessionCharacteristics);
+        } else {
+            if ($this->child->supportsSavepoints()) {
+                $this->child->transactionSavepoint("nesting" . $this->transactionNesting);
+            }
+            $this->transactionNesting++;
+        }
+    }
+
+    public function transactionEnd($chain = false)
+    {
+        if ($this->mustRollback) {
+            throw new DatabaseException("Child transaction was rolled back, so parent can't be committed");
+        }
+
+        if ($this->transactionNesting < 1) {
+            throw new DatabaseException("Not within a transaction, so can't commit");
+        }
+
+        $this->transactionNesting--;
+
+        if ($this->transactionNesting === 0) {
+            $this->child->transactionEnd();
+        }
+
+        if ($chain) {
+            return $this->transactionStart();
+        }
+    }
+
+    public function transactionRollback($savepoint = null)
+    {
+        if ($this->transactionNesting < 1) {
+            throw new DatabaseException("Not within a transaction, so can't roll back");
+        }
+
+        if ($savepoint) {
+            return $this->child->transactionRollback($savepoint);
+        }
+
+        $this->transactionNesting--;
+
+        if ($this->transactionNesting === 0) {
+            $this->child->transactionRollback();
+            $this->mustRollback = false;
+        } else {
+            if ($this->child->supportsSavepoints()) {
+                $this->child->transactionRollback("nesting" . $this->transactionNesting);
+                $this->mustRollback = false;
+
+            // Without savepoints, parent transactions must roll back if a child one has
+            } else {
+                $this->mustRollback = true;
+            }
+        }
+    }
+
+    /**
+     * Return the depth of the transaction.
+     *
+     * @return int
+     */
+    public function transactionDepth()
+    {
+        return $this->transactionNesting;
+    }
+
+    public function transactionSavepoint($savepoint)
+    {
+        return $this->child->transactionSavepoint($savepoint);
+    }
+
+    public function supportsSavepoints()
+    {
+        return $this->child->supportsSavepoints();
+    }
+}
diff --git a/src/ORM/Connect/PDOConnector.php b/src/ORM/Connect/PDOConnector.php
index 6cea5590e..85aba8d84 100644
--- a/src/ORM/Connect/PDOConnector.php
+++ b/src/ORM/Connect/PDOConnector.php
@@ -10,7 +10,7 @@ use InvalidArgumentException;
 /**
  * PDO driver database connector
  */
-class PDOConnector extends DBConnector
+class PDOConnector extends DBConnector implements TransactionManager
 {
 
     /**
@@ -21,6 +21,15 @@ class PDOConnector extends DBConnector
      */
     private static $emulate_prepare = false;
 
+    /**
+     * Should we return everything as a string in order to allow transaction savepoints?
+     * This preserves the behaviour of <= 4.3, including some bugs.
+     *
+     * @config
+     * @var boolean
+     */
+    private static $legacy_types = false;
+
     /**
      * Default strong SSL cipher to be used
      *
@@ -68,7 +77,13 @@ class PDOConnector extends DBConnector
      * Driver
      * @var string
      */
-    private $driver = null;
+    protected $driver = null;
+
+    /*
+     * Is a transaction currently active?
+     * @var bool
+     */
+    protected $inTransaction = false;
 
     /**
      * Flush all prepared statements
@@ -202,14 +217,19 @@ class PDOConnector extends DBConnector
             $options[PDO::MYSQL_ATTR_SSL_CIPHER] = array_key_exists('ssl_cipher', $parameters) ? $parameters['ssl_cipher'] : self::config()->get('ssl_cipher_default');
         }
 
-        // Set emulate prepares (unless null / default)
-        $isEmulatePrepares = self::is_emulate_prepare();
-        if (isset($isEmulatePrepares)) {
-            $options[PDO::ATTR_EMULATE_PREPARES] = (bool)$isEmulatePrepares;
-        }
+        if (static::config()->get('legacy_types')) {
+            $options[PDO::ATTR_STRINGIFY_FETCHES] = true;
+            $options[PDO::ATTR_EMULATE_PREPARES] = true;
+        } else {
+            // Set emulate prepares (unless null / default)
+            $isEmulatePrepares = self::is_emulate_prepare();
+            if (isset($isEmulatePrepares)) {
+                $options[PDO::ATTR_EMULATE_PREPARES] = (bool)$isEmulatePrepares;
+            }
 
-        // Disable stringified fetches
-        $options[PDO::ATTR_STRINGIFY_FETCHES] = false;
+            // Disable stringified fetches
+            $options[PDO::ATTR_STRINGIFY_FETCHES] = false;
+        }
 
         // May throw a PDOException if fails
         $this->pdoConnection = new PDO(
@@ -229,6 +249,8 @@ class PDOConnector extends DBConnector
     /**
      * Return the driver for this connector
      * E.g. 'mysql', 'sqlsrv', 'pgsql'
+     *
+     * @return string
      */
     public function getDriver()
     {
@@ -490,4 +512,60 @@ class PDOConnector extends DBConnector
     {
         return $this->databaseName && $this->pdoConnection;
     }
+
+    public function transactionStart($transactionMode = false, $sessionCharacteristics = false)
+    {
+        $this->inTransaction = true;
+
+        if ($transactionMode) {
+            $this->query("SET TRANSACTION $transactionMode");
+        }
+
+        if ($this->pdoConnection->beginTransaction()) {
+            if ($sessionCharacteristics) {
+                $this->query("SET SESSION CHARACTERISTICS AS TRANSACTION $sessionCharacteristics");
+            }
+            return true;
+        }
+        return false;
+    }
+
+    public function transactionEnd()
+    {
+        $this->inTransaction = false;
+        return $this->pdoConnection->commit();
+    }
+
+    public function transactionRollback($savepoint = null)
+    {
+        if ($savepoint) {
+            if ($this->supportsSavepoints()) {
+                $this->exec("ROLLBACK TO SAVEPOINT $savepoint");
+            } else {
+                throw new DatabaseException("Savepoints not supported on this PDO connection");
+            }
+        }
+
+        $this->inTransaction = false;
+        return $this->pdoConnection->rollBack();
+    }
+
+    public function transactionDepth()
+    {
+        return (int)$this->inTransaction;
+    }
+
+    public function transactionSavepoint($savepoint = null)
+    {
+        if ($this->supportsSavepoints()) {
+            $this->exec("SAVEPOINT $savepoint");
+        } else {
+            throw new DatabaseException("Savepoints not supported on this PDO connection");
+        }
+    }
+
+    public function supportsSavepoints()
+    {
+        return static::config()->get('legacy_types');
+    }
 }
diff --git a/src/ORM/Connect/TransactionManager.php b/src/ORM/Connect/TransactionManager.php
new file mode 100644
index 000000000..bec59835f
--- /dev/null
+++ b/src/ORM/Connect/TransactionManager.php
@@ -0,0 +1,59 @@
+<?php
+
+namespace SilverStripe\ORM\Connect;
+
+/**
+ * Represents an object that is capable of controlling transactions
+ */
+interface TransactionManager
+{
+    /**
+     * Start a prepared transaction
+     *
+     * @param string|boolean $transactionMode Transaction mode, or false to ignore. Deprecated and will be removed in SS5.
+     * @param string|boolean $sessionCharacteristics Session characteristics, or false to ignore. Deprecated and will be removed in SS5.
+     * @throws DatabaseException on failure
+     * @return bool True on success
+     */
+    public function transactionStart($transactionMode = false, $sessionCharacteristics = false);
+
+    /**
+     * Complete a transaction
+     *
+     * @throws DatabaseException on failure
+     * @return bool True on success
+     */
+    public function transactionEnd();
+
+    /**
+     * Roll-back a transaction
+     *
+     * @param string $savepoint If set, roll-back to the named savepoint
+     * @throws DatabaseException on failure
+     * @return bool True on success
+     */
+    public function transactionRollback($savepoint = null);
+
+    /**
+     * Create a new savepoint
+     *
+     * @param string $savepoint The savepoint name
+     * @throws DatabaseException on failure
+     */
+    public function transactionSavepoint($savepoint);
+
+    /**
+     * Return the depth of the transaction
+     * For unnested transactions returns 1 while in a transaction, 0 otherwise
+     *
+     * @return int
+     */
+    public function transactionDepth();
+
+    /**
+     * Return true if savepoints are supported by this transaction manager
+     *
+     * @return boolean
+     */
+    public function supportsSavepoints();
+}
diff --git a/tests/php/ORM/DatabaseTest.php b/tests/php/ORM/DatabaseTest.php
index d05a44fc8..247224ae5 100644
--- a/tests/php/ORM/DatabaseTest.php
+++ b/tests/php/ORM/DatabaseTest.php
@@ -21,6 +21,11 @@ class DatabaseTest extends SapphireTest
 
     protected $usesDatabase = true;
 
+    /**
+     * Disable transactions so that we can test them
+     */
+    protected $usesTransactions = false;
+
     public function testDontRequireField()
     {
         $schema = DB::get_schema();
diff --git a/tests/php/ORM/TransactionTest.php b/tests/php/ORM/TransactionTest.php
index 6a0d2b122..88617f26c 100644
--- a/tests/php/ORM/TransactionTest.php
+++ b/tests/php/ORM/TransactionTest.php
@@ -11,6 +11,8 @@ class TransactionTest extends SapphireTest
 {
     protected $usesDatabase = true;
 
+    protected $usesTransactions = false;
+
     protected static $extra_dataobjects = [
         TransactionTest\TestObject::class,
     ];
@@ -51,6 +53,7 @@ class TransactionTest extends SapphireTest
 
     public function testCreateWithTransaction()
     {
+        // First/Second in a successful transaction
         DB::get_conn()->transactionStart();
         $obj = new TransactionTest\TestObject();
         $obj->Title = 'First page';
@@ -59,10 +62,10 @@ class TransactionTest extends SapphireTest
         $obj = new TransactionTest\TestObject();
         $obj->Title = 'Second page';
         $obj->write();
+        DB::get_conn()->transactionEnd();
 
-        //Create a savepoint here:
-        DB::get_conn()->transactionSavepoint('rollback');
-
+        // Third/Fourth in a rolled back transaction
+        DB::get_conn()->transactionStart();
         $obj = new TransactionTest\TestObject();
         $obj->Title = 'Third page';
         $obj->write();
@@ -70,11 +73,8 @@ class TransactionTest extends SapphireTest
         $obj = new TransactionTest\TestObject();
         $obj->Title = 'Fourth page';
         $obj->write();
+        DB::get_conn()->transactionRollback();
 
-        //Revert to a savepoint:
-        DB::get_conn()->transactionRollback('rollback');
-
-        DB::get_conn()->transactionEnd();
 
         $first = DataObject::get(TransactionTest\TestObject::class, "\"Title\"='First page'");
         $second = DataObject::get(TransactionTest\TestObject::class, "\"Title\"='Second page'");
@@ -85,7 +85,7 @@ class TransactionTest extends SapphireTest
         $this->assertTrue(is_object($first) && $first->exists());
         $this->assertTrue(is_object($second) && $second->exists());
 
-        //These pages should NOT exist, we reverted to a savepoint:
+        //These pages should NOT exist, we rolled back
         $this->assertFalse(is_object($third) && $third->exists());
         $this->assertFalse(is_object($fourth) && $fourth->exists());
     }