[Sharding] Moved doctrine-shards into DBAL package

lib/Doctrine/DBAL/Sharding/DefaultSchemaSynchronizer.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+namespace Doctrine\DBAL\Sharding;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Schema\Comparator;
+use Doctrine\DBAL\Schema\Visitor\DropSchemaSqlCollector;
+
+/**
+ * Schema Synchronizer for Default DBAL Connection
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+class DefaultSchemaSynchronizer implements SchemaSynchronizer
+{
+    /**
+     * @var Doctrine\DBAL\Connection
+     */
+    private $conn;
+
+    /**
+     * @var Doctrine\DBAL\Platforms\AbstractPlatform
+     */
+    private $platform;
+
+    public function __construct(Connection $conn)
+    {
+        $this->conn     = $conn;
+        $this->platform = $conn->getDatabasePlatform();
+    }
+
+    /**
+     * Get the SQL statements that can be executed to create the schema.
+     *
+     * @param Schema $createSchema
+     * @return array
+     */
+    public function getCreateSchema(Schema $createSchema)
+    {
+        return $createSchema->toSql($this->platform);
+    }
+
+    /**
+     * Get the SQL Statements to update given schema with the underlying db.
+     *
+     * @param Schema $toSchema
+     * @param bool $noDrops
+     * @return array
+     */
+    public function getUpdateSchema(Schema $toSchema, $noDrops = false)
+    {
+        $comparator = new Comparator();
+        $sm         = $this->conn->getSchemaManager();
+
+        $fromSchema = $sm->createSchema();
+        $schemaDiff = $comparator->compare($fromSchema, $toSchema);
+
+        if ($noDrops) {
+            return $schemaDiff->toSaveSql($this->platform);
+        } else {
+            return $schemaDiff->toSql($this->platform);
+        }
+    }
+
+    /**
+     * Get the SQL Statements to drop the given schema from underlying db.
+     *
+     * @param Schema $dropSchema
+     * @return array
+     */
+    public function getDropSchema(Schema $dropSchema)
+    {
+        $visitor    = new DropSchemaSqlCollector($this->platform);
+        $sm         = $this->conn->getSchemaManager();
+
+        $fullSchema = $sm->createSchema();
+
+        foreach ($fullSchema->getTables() as $table) {
+            if ( $dropSchema->hasTable($table->getName())) {
+                $visitor->acceptTable($table);
+            }
+
+            foreach ($table->getForeignKeys() as $foreignKey) {
+                if ( ! $dropSchema->hasTable($table->getName())) {
+                    continue;
+                }
+
+                if ( ! $dropSchema->hasTable($foreignKey->getForeignTableName())) {
+                    continue;
+                }
+
+                $visitor->acceptForeignKey($table, $foreignKey);
+            }
+        }
+
+        if ( ! $this->platform->supportsSequences()) {
+            return $visitor->getQueries();
+        }
+
+        foreach ($dropSchema->getSequences() as $sequence) {
+            $visitor->acceptSequence($sequence);
+        }
+
+        foreach ($dropSchema->getTables() as $table) {
+            /* @var $sequence Table */
+            if ( ! $table->hasPrimaryKey()) {
+                continue;
+            }
+
+            $columns = $table->getPrimaryKey()->getColumns();
+            if (count($columns) > 1) {
+                continue;
+            }
+
+            $checkSequence = $table->getName() . "_" . $columns[0] . "_seq";
+            if ($fullSchema->hasSequence($checkSequence)) {
+                $visitor->acceptSequence($fullSchema->getSequence($checkSequence));
+            }
+        }
+
+        return $visitor->getQueries();
+    }
+
+    /**
+     * Get the SQL statements to drop all schema assets from underlying db.
+     *
+     * @return array
+     */
+    public function getDropAllSchema()
+    {
+        $sm      = $this->conn->getSchemaManager();
+        $visitor = new \Doctrine\DBAL\Schema\Visitor\DropSchemaSqlCollector($this->platform);
+
+        /* @var $schema \Doctrine\DBAL\Schema\Schema */
+        $schema  = $sm->createSchema();
+        $schema->visit($visitor);
+
+        return $visitor->getQueries();
+    }
+
+    /**
+     * Create the Schema
+     *
+     * @param Schema $createSchema
+     * @return void
+     */
+    public function createSchema(Schema $createSchema)
+    {
+        $this->processSql($this->getCreateSchema($createSchema));
+    }
+
+    /**
+     * Update the Schema to new schema version.
+     *
+     * @param Schema $toSchema
+     * @param bool $noDrops
+     * @return void
+     */
+    public function updateSchema(Schema $toSchema, $noDrops = false)
+    {
+        $this->processSql($this->getUpdateSchema($toSchema, $noDrops));
+    }
+
+    /**
+     * Drop the given database schema from the underlying db.
+     *
+     * @param Schema $dropSchema
+     * @return void
+     */
+    public function dropSchema(Schema $dropSchema)
+    {
+        $this->processSqlSafely($this->getDropSchema($dropSchema));
+    }
+
+    /**
+     * Drop all assets from the underyling db.
+     *
+     * @return void
+     */
+    public function dropAllSchema()
+    {
+        $this->processSql($this->getDropAllSchema());
+    }
+
+    private function processSqlSafely(array $sql)
+    {
+        foreach ($sql as $s) {
+            try {
+                $this->conn->exec($s);
+            } catch(\Exception $e) {
+
+            }
+        }
+    }
+
+    private function processSql(array $sql)
+    {
+        foreach ($sql as $s) {
+            $this->conn->exec($s);
+        }
+    }
+}
+

lib/Doctrine/DBAL/Sharding/PoolingShardConnection.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Event\ConnectionEventArgs;
+use Doctrine\DBAL\Events;
+use Doctrine\DBAL\Driver;
+use Doctrine\DBAL\Configuration;
+
+use Doctrine\Common\EventManager;
+
+use Doctrine\DBAL\Sharding\ShardChoser\ShardChoser;
+
+/**
+ * Sharding implementation that pools many different connections
+ * internally and serves data from the currently active connection.
+ *
+ * The internals of this class are:
+ *
+ * - All sharding clients are specified and given a shard-id during
+ *   configuration.
+ * - By default, the global shard is selected. If no global shard is configured
+ *   an exception is thrown on access.
+ * - Selecting a shard by distribution value delegates the mapping
+ *   "distributionValue" => "client" to the ShardChooser interface.
+ * - An exception is thrown if trying to switch shards during an open
+ *   transaction.
+ *
+ * Instantiation through the DriverManager looks like:
+ *
+ * @example
+ *
+ * $conn = DriverManager::getConnection(array(
+ *    'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+ *    'driver' => 'pdo_mysql',
+ *    'global' => array('user' => '', 'password' => '', 'host' => '', 'dbname' => ''),
+ *    'shards' => array(
+ *        array('id' => 1, 'user' => 'slave1', 'password', 'host' => '', 'dbname' => ''),
+ *        array('id' => 2, 'user' => 'slave2', 'password', 'host' => '', 'dbname' => ''),
+ *    ),
+ *    'shardChoser' => 'Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser',
+ * ));
+ * $shardManager = $conn->getShardManager();
+ * $shardManager->selectGlobal();
+ * $shardManager->selectShard($value);
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+class PoolingShardConnection extends Connection
+{
+    /**
+     * @var array
+     */
+    private $activeConnections;
+
+    /**
+     * @var int
+     */
+    private $activeShardId;
+
+    /**
+     * @var array
+     */
+    private $connections;
+
+    /**
+     * @var PoolingShardManager
+     */
+    private $shardManager;
+
+    public function __construct(array $params, Driver $driver, Configuration $config = null, EventManager $eventManager = null)
+    {
+        if ( !isset($params['global']) || !isset($params['shards'])) {
+            throw new \InvalidArgumentException("Connection Parameters require 'global' and 'shards' configurations.");
+        }
+
+        if ( !isset($params['shardChoser'])) {
+            throw new \InvalidArgumentException("Missing Shard Choser configuration 'shardChoser'");
+        }
+
+        if (is_string($params['shardChoser'])) {
+            $params['shardChoser'] = new $params['shardChoser'];
+        }
+
+        if ( ! ($params['shardChoser'] instanceof ShardChoser)) {
+            throw new \InvalidArgumentException("The 'shardChoser' configuration is not a valid instance of Doctrine\DBAL\Sharding\ShardChoser\ShardChoser"); 
+        }
+
+        $this->connections[0] = array_merge($params, $params['global']);
+
+        foreach ($params['shards'] as $shard) {
+            if ( ! isset($shard['id'])) {
+                throw new \InvalidArgumentException("Missing 'id' for one configured shard. Please specificy a unique shard-id.");
+            }
+
+            if ( !is_numeric($shard['id']) || $shard['id'] < 1) {
+                throw new \InvalidArgumentException("Shard Id has to be a non-negative number.");
+            }
+
+            if (isset($this->connections[$shard['id']])) {
+                throw new \InvalidArgumentException("Shard " . $shard['id'] . " is duplicated in the configuration.");
+            }
+
+            $this->connections[$shard['id']] = array_merge($params, $shard);
+        }
+
+        parent::__construct($params, $driver, $config, $eventManager);
+    }
+
+    /**
+     * @return \Doctrine\DBAL\Sharding\PoolingShardManager
+     */
+    public function getShardManager()
+    {
+        if ($this->shardManager === null) {
+            $params = $this->getParams();
+            $this->shardManager = new PoolingShardManager($this, $params['shardChoser']);
+        }
+        return $this->shardManager;
+    }
+
+    public function connect($shardId = null)
+    {
+        if ($shardId === null && $this->_conn) {
+            return false;
+        }
+
+        if ($shardId !== null && $shardId === $this->activeShardId) {
+            return false;
+        }
+
+        if ($this->getTransactionNestingLevel() > 0) {
+            throw new ShardingException("Cannot switch shard when transaction is active.");
+        }
+
+        $this->activeShardId = (int)$shardId;
+
+        if (isset($this->activeConnections[$this->activeShardId])) {
+            $this->_conn = $this->activeConnections[$this->activeShardId];
+            return false;
+        }
+
+        $this->_conn = $this->activeConnections[$this->activeShardId] = $this->connectTo($this->activeShardId);
+
+        if ($this->_eventManager->hasListeners(Events::postConnect)) {
+            $eventArgs = new Event\ConnectionEventArgs($this);
+            $this->_eventManager->dispatchEvent(Events::postConnect, $eventArgs);
+        }
+
+        return true;
+    }
+
+
+    /**
+     * Connect to a specific connection
+     *
+     * @param  string $shardId
+     * @return Driver
+     */
+    protected function connectTo($shardId)
+    {
+        $params = $this->getParams();
+
+        $driverOptions = isset($params['driverOptions']) ? $params['driverOptions'] : array();
+
+        $connectionParams = $this->connections[$shardId];
+
+        $user = isset($connectionParams['user']) ? $connectionParams['user'] : null;
+        $password = isset($connectionParams['password']) ? $connectionParams['password'] : null;
+
+        return $this->_driver->connect($connectionParams, $user, $password, $driverOptions);
+    }
+
+    public function isConnected($shardId = null)
+    {
+        if ($shardId === null) {
+            return ($this->_conn !== null);
+        }
+
+        return isset($this->activeConnections[$shardId]);
+    }
+
+    public function close()
+    {
+        unset($this->_conn);
+        unset($this->activeConnections);
+    }
+}
+

lib/Doctrine/DBAL/Sharding/PoolingShardManager.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding;
+
+use Doctrine\DBAL\Sharding\ShardChoser\ShardChoser;
+
+/**
+ * Shard Manager for the Connection Pooling Shard Strategy
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+class PoolingShardManager implements ShardManager
+{
+    private $conn;
+    private $choser;
+    private $currentDistributionValue;
+
+    public function __construct(PoolingShardConnection $conn, ShardChoser $choser)
+    {
+        $this->conn = $conn;
+        $this->choser = $choser;
+    }
+
+    public function selectGlobal()
+    {
+        $this->conn->connect(0);
+        $this->currentDistributionValue = null;
+    }
+
+    public function selectShard($distributionValue)
+    {
+        $shardId = $this->choser->pickShard($distributionValue, $this->conn);
+        $this->conn->connect($shardId);
+        $this->currentDistributionValue = $distributionValue;
+    }
+
+    public function getCurrentDistributionValue()
+    {
+        return $this->currentDistributionValue;
+    }
+
+    public function getShards()
+    {
+        $params = $this->conn->getParams();
+        $shards = array();
+
+        foreach ($params['shards'] as $shard) {
+            $shards[] = array('id' => $shard['id']);
+        }
+
+        return $shards;
+    }
+
+    public function queryAll($sql, array $params, array $types)
+    {
+        $shards = $this->getShards();
+        if (!$shards) {
+            throw new \RuntimeException("No shards found.");
+        }
+
+        $result = array();
+        $oldDistribution = $this->getCurrentDistributionValue();
+
+        foreach ($shards as $shard) {
+            $this->selectShard($shard['id']);
+            foreach ($this->conn->fetchAll($sql, $params, $types) as $row) {
+                $result[] = $row;
+            }
+        }
+
+        if ($oldDistribution === null) {
+            $this->selectGlobal();
+        } else {
+            $this->selectShard($oldDistribution);
+        }
+
+        return $result;
+    }
+}
+

lib/Doctrine/DBAL/Sharding/SQLAzure/SQLAzureSchemaSynchronizer.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding\SQLAzure;
+
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Types\Type;
+
+use Doctrine\DBAL\Sharding\SchemaSynchronizer;
+use Doctrine\DBAL\Sharding\DefaultSchemaSynchronizer;
+
+/**
+ * SQL Azure Schema Synchronizer
+ *
+ * Will iterate over all shards when performing schema operations. This is done
+ * by partioning the passed schema into subschemas for the federation and the
+ * global database and then applying the operations step by step using the
+ * {@see \Doctrine\DBAL\Sharding\DefaultSchemaSynchronizer}.
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+class SQLAzureSchemaSynchronizer implements SchemaSynchronizer
+{
+    /**
+     * @var Connection
+     */
+    private $conn;
+
+    /**
+     * @var SQLAzureShardManager
+     */
+    private $shardManager;
+
+    /**
+     * @var SchemaSynchronizer
+     */
+    private $synchronizer;
+
+    const FEDERATION_TABLE_FEDERATED   = 'azure.federated';
+    const FEDERATION_DISTRIBUTION_NAME = 'azure.federatedOnDistributionName';
+
+    public function __construct(Connection $conn, SQLAzureShardManager $shardManager, SchemaSynchronizer $sync = null)
+    {
+        $this->conn = $conn;
+        $this->shardManager = $shardManager;
+        $this->synchronizer = $sync ?: new DefaultSchemaSynchronizer($conn);
+    }
+
+
+    /**
+     * Get the SQL statements that can be executed to create the schema.
+     *
+     * @param Schema $createSchema
+     * @return array
+     */
+    public function getCreateSchema(Schema $createSchema)
+    {
+        $sql = array();
+
+        list($global, $federation) = $this->partitionSchema($createSchema);
+
+        $globalSql = $this->synchronizer->getCreateSchema($global);
+        if ($globalSql) {
+            $sql[] = "-- Create Root Federation\n" .
+                     "USE FEDERATION ROOT WITH RESET;";
+            $sql = array_merge($sql, $globalSql);
+        }
+
+        $federationSql = $this->synchronizer->getCreateSchema($federation);
+
+        if ($federationSql) {
+            $defaultValue = $this->getFederationTypeDefaultValue();
+
+            $sql[] = $this->getCreateFederationStatement();
+            $sql[] = "USE FEDERATION " . $this->shardManager->getFederationName() . " (" . $this->shardManager->getDistributionKey() . " = " . $defaultValue . ") WITH RESET, FILTERING = OFF;";
+            $sql = array_merge($sql, $federationSql);
+        }
+
+        return $sql;
+    }
+
+    /**
+     * Get the SQL Statements to update given schema with the underlying db.
+     *
+     * @param Schema $toSchema
+     * @param bool $noDrops
+     * @return array
+     */
+    public function getUpdateSchema(Schema $toSchema, $noDrops = false)
+    {
+        return $this->work($toSchema, function($synchronizer, $schema) use ($noDrops) {
+            return $synchronizer->getUpdateSchema($schema, $noDrops);
+        });
+    }
+
+    /**
+     * Get the SQL Statements to drop the given schema from underlying db.
+     *
+     * @param Schema $dropSchema
+     * @return array
+     */
+    public function getDropSchema(Schema $dropSchema)
+    {
+        return $this->work($dropSchema, function($synchronizer, $schema) {
+            return $synchronizer->getDropSchema($schema);
+        });
+    }
+
+    /**
+     * Create the Schema
+     *
+     * @param Schema $createSchema
+     * @return void
+     */
+    public function createSchema(Schema $createSchema)
+    {
+        $this->processSql($this->getCreateSchema($createSchema));
+    }
+
+    /**
+     * Update the Schema to new schema version.
+     *
+     * @param Schema $toSchema
+     * @return void
+     */
+    public function updateSchema(Schema $toSchema, $noDrops = false)
+    {
+        $this->processSql($this->getUpdateSchema($toSchema, $noDrops));
+    }
+
+    /**
+     * Drop the given database schema from the underlying db.
+     *
+     * @param Schema $dropSchema
+     * @return void
+     */
+    public function dropSchema(Schema $dropSchema)
+    {
+        $this->processSqlSafely($this->getDropSchema($dropSchema));
+    }
+
+    /**
+     * Get the SQL statements to drop all schema assets from underlying db.
+     *
+     * @return array
+     */
+    public function getDropAllSchema()
+    {
+        $this->shardManager->selectGlobal();
+        $globalSql = $this->synchronizer->getDropAllSchema();
+
+        if ($globalSql) {
+            $sql[] = "-- Work on Root Federation\nUSE FEDERATION ROOT WITH RESET;";
+            $sql = array_merge($sql, $globalSql);
+        }
+
+        $shards = $this->shardManager->getShards();
+        foreach ($shards as $shard) {
+            $this->shardManager->selectShard($shard['rangeLow']);
+
+            $federationSql = $this->synchronizer->getDropAllSchema();
+            if ($federationSql) {
+                $sql[] = "-- Work on Federation ID " . $shard['id'] . "\n" .
+                         "USE FEDERATION " . $this->shardManager->getFederationName() . " (" . $this->shardManager->getDistributionKey() . " = " . $shard['rangeLow'].") WITH RESET, FILTERING = OFF;";
+                $sql = array_merge($sql, $federationSql);
+            }
+        }
+
+        $sql[] = "USE FEDERATION ROOT WITH RESET;";
+        $sql[] = "DROP FEDERATION " . $this->shardManager->getFederationName();
+
+        return $sql;
+    }
+
+    /**
+     * Drop all assets from the underyling db.
+     *
+     * @return void
+     */
+    public function dropAllSchema()
+    {
+        $this->processSqlSafely($this->getDropAllSchema());
+    }
+
+    private function partitionSchema(Schema $schema)
+    {
+        return array(
+            $this->extractSchemaFederation($schema, false),
+            $this->extractSchemaFederation($schema, true),
+        );
+    }
+
+    private function extractSchemaFederation(Schema $schema, $isFederation)
+    {
+        $partionedSchema = clone $schema;
+
+        foreach ($partionedSchema->getTables() as $table) {
+            if ($isFederation) {
+                $table->addOption(self::FEDERATION_DISTRIBUTION_NAME, $this->shardManager->getDistributionKey());
+            }
+
+            if ( $table->hasOption(self::FEDERATION_TABLE_FEDERATED) !== $isFederation) {
+                $partionedSchema->dropTable($table->getName());
+            } else {
+                foreach ($table->getForeignKeys() as $fk) {
+                    $foreignTable = $schema->getTable($fk->getForeignTableName());
+                    if ($foreignTable->hasOption(self::FEDERATION_TABLE_FEDERATED) !== $isFederation) {
+                        throw new \RuntimeException("Cannot have foreign key between global/federation.");
+                    }
+                }
+            }
+        }
+
+        return $partionedSchema;
+    }
+
+    private function processSqlSafely(array $sql)
+    {
+        foreach ($sql as $s) {
+            try {
+                $this->conn->exec($s);
+            } catch(\Exception $e) {
+
+            }
+        }
+    }
+
+    private function processSql(array $sql)
+    {
+        foreach ($sql as $s) {
+            $this->conn->exec($s);
+        }
+    }
+
+    /**
+     * Work on the Global/Federation based on currently existing shards and
+     * perform the given operation on the underyling schema synchronizer given
+     * the different partioned schema instances.
+     *
+     * @param Schema $schema
+     * @param Closure $operation
+     * @return array
+     */
+    private function work(Schema $schema, \Closure $operation)
+    {
+        list($global, $federation) = $this->partitionSchema($schema);
+        $sql = array();
+
+        $this->shardManager->selectGlobal();
+        $globalSql = $operation($this->synchronizer, $global);
+
+        if ($globalSql) {
+            $sql[] = "-- Work on Root Federation\nUSE FEDERATION ROOT WITH RESET;";
+            $sql   = array_merge($sql, $globalSql);
+        }
+
+        $shards = $this->shardManager->getShards();
+
+        foreach ($shards as $shard) {
+            $this->shardManager->selectShard($shard['rangeLow']);
+
+            $federationSql = $operation($this->synchronizer, $federation);
+            if ($federationSql) {
+                $sql[] = "-- Work on Federation ID " . $shard['id'] . "\n" .
+                         "USE FEDERATION " . $this->shardManager->getFederationName() . " (" . $this->shardManager->getDistributionKey() . " = " . $shard['rangeLow'].") WITH RESET, FILTERING = OFF;";
+                $sql   = array_merge($sql, $federationSql);
+            }
+        }
+
+        return $sql;
+    }
+
+    private function getFederationTypeDefaultValue()
+    {
+        $federationType = Type::getType($this->shardManager->getDistributionType());
+
+        switch ($federationType->getName()) {
+            case Type::GUID:
+                $defaultValue = '00000000-0000-0000-0000-000000000000';
+                break;
+            case Type::INTEGER:
+            case Type::SMALLINT:
+            case Type::BIGINT:
+                $defaultValue = '0';
+                break;
+            default:
+                $defaultValue = '';
+                break;
+        }
+        return $defaultValue;
+    }
+
+    private function getCreateFederationStatement()
+    {
+        $federationType = Type::getType($this->shardManager->getDistributionType());
+        $federationTypeSql = $federationType->getSqlDeclaration(array(), $this->conn->getDatabasePlatform());
+
+        return "--Create Federation\n" .
+               "CREATE FEDERATION " . $this->shardManager->getFederationName() . " (" . $this->shardManager->getDistributionKey() . " " . $federationTypeSql ."  RANGE)";
+    }
+}
+

lib/Doctrine/DBAL/Sharding/SQLAzure/SQLAzureShardManager.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding\SQLAzure;
+
+use Doctrine\DBAL\Sharding\ShardManager;
+use Doctrine\DBAL\Sharding\ShardingException;
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Types\Type;
+
+/**
+ * Sharding using the SQL Azure Federations support.
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+class SQLAzureShardManager implements ShardManager
+{
+    /**
+     * @var string
+     */
+    private $federationName;
+
+    /**
+     * @var bool
+     */
+    private $filteringEnabled;
+
+    /**
+     * @var string
+     */
+    private $distributionKey;
+
+    /**
+     * @var string
+     */
+    private $distributionType;
+
+    /**
+     * @var Connection
+     */
+    private $conn;
+
+    /**
+     * @var string
+     */
+    private $currentDistributionValue;
+
+    /**
+     * @param Connection $conn
+     */
+    public function __construct(Connection $conn)
+    {
+        $this->conn = $conn;
+        $params = $conn->getParams();
+
+        if ( ! isset($params['sharding']['federationName'])) {
+            throw ShardingException::missingDefaultFederationName();
+        }
+
+        if ( ! isset($params['sharding']['distributionKey'])) {
+            throw ShardingException::missingDefaultDistributionKey();
+        }
+
+        if ( ! isset($params['sharding']['distributionType'])) {
+            throw ShardingException::missingDistributionType();
+        }
+
+        $this->federationName = $params['sharding']['federationName'];
+        $this->distributionKey = $params['sharding']['distributionKey'];
+        $this->distributionType = $params['sharding']['distributionType'];
+        $this->filteringEnabled = (isset($params['sharding']['filteringEnabled'])) ? (bool)$params['sharding']['filteringEnabled'] : false;
+    }
+
+    /**
+     * Get name of the federation
+     *
+     * @return string
+     */
+    public function getFederationName()
+    {
+        return $this->federationName;
+    }
+
+    /**
+     * Get the distribution key
+     *
+     * @return string
+     */
+    public function getDistributionKey()
+    {
+        return $this->distributionKey;
+    }
+
+    /**
+     * Get the Doctrine Type name used for the distribution
+     *
+     * @return string
+     */
+    public function getDistributionType()
+    {
+        return $this->distributionType;
+    }
+
+    /**
+     * Enabled/Disable filtering on the fly.
+     *
+     * @param bool $flag
+     * @return void
+     */
+    public function setFilteringEnabled($flag)
+    {
+        $this->filteringEnabled = (bool)$flag;
+    }
+
+    /**
+     * @override
+     * {@inheritDoc}
+     */
+    public function selectGlobal()
+    {
+        if ($this->conn->isTransactionActive()) {
+            throw ShardingException::activeTransaction();
+        }
+
+        $sql = "USE FEDERATION ROOT WITH RESET";
+        $this->conn->exec($sql);
+        $this->currentDistributionValue = null;
+    }
+
+    /**
+     * @override
+     * {@inheritDoc}
+     */
+    public function selectShard($distributionValue)
+    {
+        if ($this->conn->isTransactionActive()) {
+            throw ShardingException::activeTransaction();
+        }
+
+        if ($distributionValue === null || is_bool($distributionValue) || !is_scalar($distributionValue)) {
+            throw ShardingException::noShardDistributionValue();
+        }
+
+        $platform = $this->conn->getDatabasePlatform();
+        $sql = sprintf(
+            "USE FEDERATION %s (%s = %s) WITH RESET, FILTERING = %s;",
+            $platform->quoteIdentifier($this->federationName),
+            $platform->quoteIdentifier($this->distributionKey),
+            $this->conn->quote($distributionValue),
+            ($this->filteringEnabled ? 'ON' : 'OFF')
+        );
+
+        $this->conn->exec($sql);
+        $this->currentDistributionValue = $distributionValue;
+    }
+
+    /**
+     * @override
+     * {@inheritDoc}
+     */
+    public function getCurrentDistributionValue()
+    {
+        return $this->currentDistributionValue;
+    }
+
+    /**
+     * @override
+     * {@inheritDoc}
+     */
+    public function getShards()
+    {
+        $sql = "SELECT member_id as id,
+                      distribution_name as distribution_key,
+                      CAST(range_low AS CHAR) AS rangeLow,
+                      CAST(range_high AS CHAR) AS rangeHigh
+                      FROM sys.federation_member_distributions d
+                      INNER JOIN sys.federations f ON f.federation_id = d.federation_id
+                      WHERE f.name = " . $this->conn->quote($this->federationName);
+        return $this->conn->fetchAll($sql);
+    }
+
+     /**
+      * @override
+      * {@inheritDoc}
+      */
+    public function queryAll($sql, array $params = array(), array $types = array())
+    {
+        $shards = $this->getShards();
+        if (!$shards) {
+            throw new \RuntimeException("No shards found for " . $this->federationName);
+        }
+
+        $result = array();
+        $oldDistribution = $this->getCurrentDistributionValue();
+
+        foreach ($shards as $shard) {
+            $this->selectShard($shard['rangeLow']);
+            foreach ($this->conn->fetchAll($sql, $params, $types) as $row) {
+                $result[] = $row;
+            }
+        }
+
+        if ($oldDistribution === null) {
+            $this->selectGlobal();
+        } else {
+            $this->selectShard($oldDistribution);
+        }
+
+        return $result;
+    }
+
+    /**
+     * Split Federation at a given distribution value.
+     *
+     * @param mixed $splitDistributionValue
+     */
+    public function splitFederation($splitDistributionValue)
+    {
+        $type = Type::getType($this->distributionType);
+
+        $sql = "ALTER FEDERATION " . $this->getFederationName() . " " .
+               "SPLIT AT (" . $this->getDistributionKey() . " = " .
+               $this->conn->quote($splitDistributionValue, $type->getBindingType()) . ")";
+        $this->conn->exec($sql);
+    }
+}
+

lib/Doctrine/DBAL/Sharding/SQLAzure/Schema/MultiTenantVisitor.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding\SQLAzure\Schema;
+
+use Doctrine\DBAL\Schema\Visitor\Visitor,
+    Doctrine\DBAL\Schema\Table,
+    Doctrine\DBAL\Schema\Schema,
+    Doctrine\DBAL\Schema\Column,
+    Doctrine\DBAL\Schema\ForeignKeyConstraint,
+    Doctrine\DBAL\Schema\Constraint,
+    Doctrine\DBAL\Schema\Sequence,
+    Doctrine\DBAL\Schema\Index;
+
+/**
+ * Converts a single tenant schema into a multi-tenant schema for SQL Azure
+ * Federations under the following assumptions:
+ *
+ * - Every table is part of the multi-tenant application, only explicitly
+ *   excluded tables are non-federated. The behavior of the tables being in
+ *   global or federated database is undefined. It depends on you selecting a
+ *   federation before DDL statements or not.
+ * - Every Primary key of a federated table is extended by another column
+ *   'tenant_id' with a default value of the SQLAzure function
+ *   `federation_filtering_value('tenant_id')`.
+ * - You always have to work with `filtering=On` when using federations with this
+ *   multi-tenant approach.
+ * - Primary keys are either using globally unique ids (GUID, Table Generator)
+ *   or you explicitly add the tenent_id in every UPDATE or DELETE statement
+ *   (otherwise they will affect the same-id rows from other tenents as well).
+ *   SQLAzure throws errors when you try to create IDENTIY columns on federated
+ *   tables.
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+class MultiTenantVisitor implements Visitor
+{
+    /**
+     * @var array
+     */
+    private $excludedTables = array();
+
+    /**
+     * @var string
+     */
+    private $tenantColumnName;
+
+    /**
+     * @var string
+     */
+    private $tenantColumnType = 'integer';
+
+    /**
+     * Name of the federation distribution, defaulting to the tenantColumnName
+     * if not specified.
+     *
+     * @var string
+     */
+    private $distributionName;
+
+    public function __construct(array $excludedTables = array(), $tenantColumnName = 'tenant_id', $distributionName = null)
+    {
+        $this->excludedTables = $excludedTables;
+        $this->tenantColumnName = $tenantColumnName;
+        $this->distributionName = $distributionName ?: $tenantColumnName;
+    }
+
+    /**
+     * @param Table $table
+     */
+    public function acceptTable(Table $table)
+    {
+        if (in_array($table->getName(), $this->excludedTables)) {
+            return;
+        }
+
+        $table->addColumn($this->tenantColumnName, $this->tenantColumnType, array(
+            'default' => "federation_filtering_value('". $this->distributionName ."')",
+        ));
+
+        $clusteredIndex = $this->getClusteredIndex($table);
+
+        $indexColumns = $clusteredIndex->getColumns();
+        $indexColumns[] = $this->tenantColumnName;
+
+        if ($clusteredIndex->isPrimary()) {
+            $table->dropPrimaryKey();
+            $table->setPrimaryKey($indexColumns);
+        } else {
+            $table->dropIndex($clusteredIndex->getName());
+            $table->addIndex($indexColumns, $clusteredIndex->getName());
+            $table->getIndex($clusteredIndex->getName())->addFlag('clustered');
+        }
+    }
+
+    private function getClusteredIndex($table)
+    {
+        foreach ($table->getIndexes() as $index) {
+            if ($index->isPrimary() && ! $index->hasFlag('nonclustered')) {
+                return $index;
+            } else if ($index->hasFlag('clustered')) {
+                return $index;
+            }
+        }
+        throw new \RuntimeException("No clustered index found on table " . $table->getName());
+    }
+
+    /**
+     * @param Schema $schema
+     */
+    public function acceptSchema(Schema $schema)
+    {
+    }
+
+    /**
+     * @param Column $column
+     */
+    public function acceptColumn(Table $table, Column $column)
+    {
+    }
+
+    /**
+     * @param Table $localTable
+     * @param ForeignKeyConstraint $fkConstraint
+     */
+    public function acceptForeignKey(Table $localTable, ForeignKeyConstraint $fkConstraint)
+    {
+    }
+
+    /**
+     * @param Table $table
+     * @param Index $index
+     */
+    public function acceptIndex(Table $table, Index $index)
+    {
+    }
+
+    /**
+     * @param Sequence $sequence
+     */
+    public function acceptSequence(Sequence $sequence)
+    {
+    }
+}
+

lib/Doctrine/DBAL/Sharding/SchemaSynchronizer.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding;
+
+use Doctrine\DBAL\Schema\Schema;
+
+/**
+ * The synchronizer knows how to synchronize a schema with the configured
+ * database.
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+interface SchemaSynchronizer
+{
+    /**
+     * Get the SQL statements that can be executed to create the schema.
+     *
+     * @param Schema $createSchema
+     * @return array
+     */
+    function getCreateSchema(Schema $createSchema);
+
+    /**
+     * Get the SQL Statements to update given schema with the underlying db.
+     *
+     * @param Schema $toSchema
+     * @param bool $noDrops
+     * @return array
+     */
+    function getUpdateSchema(Schema $toSchema, $noDrops = false);
+
+    /**
+     * Get the SQL Statements to drop the given schema from underlying db.
+     *
+     * @param Schema $dropSchema
+     * @return array
+     */
+    function getDropSchema(Schema $dropSchema);
+
+    /**
+     * Get the SQL statements to drop all schema assets from underlying db.
+     *
+     * @return array
+     */
+    function getDropAllSchema();
+
+    /**
+     * Create the Schema
+     *
+     * @param Schema $createSchema
+     * @return void
+     */
+    function createSchema(Schema $createSchema);
+
+    /**
+     * Update the Schema to new schema version.
+     *
+     * @param Schema $toSchema
+     * @param bool $noDrops
+     * @return void
+     */
+    function updateSchema(Schema $toSchema, $noDrops = false);
+
+    /**
+     * Drop the given database schema from the underlying db.
+     *
+     * @param Schema $dropSchema
+     * @return void
+     */
+    function dropSchema(Schema $dropSchema);
+
+    /**
+     * Drop all assets from the underyling db.
+     *
+     * @return void
+     */
+    function dropAllSchema();
+}
+

lib/Doctrine/DBAL/Sharding/ShardChoser/MultiTenantShardChoser.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding\ShardChoser;
+
+use Doctrine\DBAL\Sharding\PoolingShardConnection;
+
+/**
+ * The MultiTenant Shard choser assumes that the distribution value directly
+ * maps to the shard id.
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+class MultiTenantShardChoser implements ShardChoser
+{
+    public function pickShard($distributionValue, PoolingShardConnection $conn)
+    {
+        return $distributionValue;
+    }
+}
+

lib/Doctrine/DBAL/Sharding/ShardChoser/ShardChoser.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding\ShardChoser;
+
+use Doctrine\DBAL\Sharding\PoolingShardConnection;
+
+/**
+ * Given a distribution value this shard-choser strategy will pick the shard to
+ * connect to for retrieving rows with the distribution value.
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+interface ShardChoser
+{
+    /**
+     * Pick a shard for the given distribution value
+     *
+     * @param string $distributionValue
+     * @param PoolingShardConnection $conn
+     * @return int
+     */
+    function pickShard($distributionValue, PoolingShardConnection $conn);
+}
+

lib/Doctrine/DBAL/Sharding/ShardManager.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding;
+
+use Doctrine\DBAL\Connection;
+
+/**
+ * Sharding Manager gives access to APIs to implementing sharding on top of
+ * Doctrine\DBAL\Connection instances.
+ *
+ * For simplicity and developer ease-of-use (and understanding) the sharding
+ * API only covers single shard queries, no fan-out support. It is primarily
+ * suited for multi-tenant applications.
+ *
+ * The assumption about sharding here
+ * is that a distribution value can be found that gives access to all the
+ * necessary data for all use-cases. Switching between shards should be done with
+ * caution, especially if lazy loading is implemented. Any query is always
+ * executed against the last shard that was selected. If a query is created for
+ * a shard Y but then a shard X is selected when its actually excecuted you
+ * will hit the wrong shard.
+ *
+ * @author Benjamin Eberlei <kontakt@beberlei.de>
+ */
+interface ShardManager
+{
+    /**
+     * Select global database with global data.
+     *
+     * This is the default database that is connected when no shard is
+     * selected.
+     *
+     * @return void
+     */
+    function selectGlobal();
+
+    /**
+     * SELECT queries after this statement will be issued against the selected
+     * shard.
+     *
+     * @throws ShardingException If no value is passed as shard identifier.
+     * @param mixed $distributionValue
+     * @param array $options
+     * @return void
+     */
+    function selectShard($distributionValue);
+
+    /**
+     * Get the distribution value currently used for sharding.
+     *
+     * @return string
+     */
+    function getCurrentDistributionValue();
+
+    /**
+     * Get information about the amount of shards and other details.
+     *
+     * Format is implementation specific, each shard is one element and has a
+     * 'name' attribute at least.
+     *
+     * @return array
+     */
+    function getShards();
+
+    /**
+     * Query all shards in undefined order and return the results appended to
+     * each other. Restore the previous distribution value after execution.
+     *
+     * Using {@link Connection::fetchAll} to retrieve rows internally.
+     *
+     * @param string $sql
+     * @param array $params
+     * @param array $types
+     * @return array
+     */
+    function queryAll($sql, array $params, array $types);
+}
+

lib/Doctrine/DBAL/Sharding/ShardingException.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\DBAL\Sharding;
+
+use Doctrine\DBAL\DBALException;
+
+/**
+ * Sharding related Exceptions
+ *
+ * @since 2.3
+ */
+class ShardingException extends DBALException
+{
+    static public function notImplemented()
+    {
+        return new self("This functionality is not implemented with this sharding provider.", 1331557937);
+    }
+
+    static public function missingDefaultFederationName()
+    {
+        return new self("SQLAzure requires a federation name to be set during sharding configuration.", 1332141280);
+    }
+
+    static public function missingDefaultDistributionKey()
+    {
+        return new self("SQLAzure requires a distribution key to be set during sharding configuration.", 1332141329);
+    }
+
+    static public function activeTransaction()
+    {
+        return new self("Cannot switch shard during an active transaction.", 1332141766);
+    }
+
+    static public function noShardDistributionValue()
+    {
+        return new self("You have to specify a string or integer as shard distribution value.", 1332142103);
+    }
+
+    static public function missingDistributionType()
+    {
+        return new self("You have to specify a sharding distribution type such as 'integer', 'string', 'guid'.");
+    }
+}
+

tests/Doctrine/Tests/DBAL/Sharding/DefaultSchemaSynchronizerTest.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\Tests\DBAL\Sharding;
+
+use Doctrine\DBAL\DriverManager;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Sharding\DefaultSchemaSynchronizer;
+
+class DefaultSchemaSynchronizerTest extends \PHPUnit_Framework_TestCase
+{
+    private $conn;
+    private $synchronizer;
+
+    public function setUp()
+    {
+        $this->conn = DriverManager::getConnection(array(
+            'driver' => 'pdo_sqlite',
+            'memory' => true,
+        ));
+        $this->synchronizer = new DefaultSchemaSynchronizer($this->conn);
+    }
+
+    public function testGetCreateSchema()
+    {
+        $schema = new Schema();
+        $table = $schema->createTable('test');
+        $table->addColumn('id', 'integer');
+        $table->setPrimaryKey(array('id'));
+
+        $sql = $this->synchronizer->getCreateSchema($schema);
+        $this->assertEquals(array('CREATE TABLE test (id INTEGER NOT NULL, PRIMARY KEY("id"))'), $sql);
+    }
+
+    public function testGetUpdateSchema()
+    {
+        $schema = new Schema();
+        $table = $schema->createTable('test');
+        $table->addColumn('id', 'integer');
+        $table->setPrimaryKey(array('id'));
+
+        $sql = $this->synchronizer->getUpdateSchema($schema);
+        $this->assertEquals(array('CREATE TABLE test (id INTEGER NOT NULL, PRIMARY KEY("id"))'), $sql);
+    }
+
+    public function testGetDropSchema()
+    {
+        $schema = new Schema();
+        $table = $schema->createTable('test');
+        $table->addColumn('id', 'integer');
+        $table->setPrimaryKey(array('id'));
+
+        $this->synchronizer->createSchema($schema);
+
+        $sql = $this->synchronizer->getDropSchema($schema);
+        $this->assertEquals(array('DROP TABLE test'), $sql);
+    }
+
+    public function testGetDropAllSchema()
+    {
+        $schema = new Schema();
+        $table = $schema->createTable('test');
+        $table->addColumn('id', 'integer');
+        $table->setPrimaryKey(array('id'));
+
+        $this->synchronizer->createSchema($schema);
+
+        $sql = $this->synchronizer->getDropAllSchema();
+        $this->assertEquals(array('DROP TABLE test'), $sql);
+    }
+}
+

tests/Doctrine/Tests/DBAL/Sharding/PoolingShardConnectionTest.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\Tests\DBAL\Sharding;
+
+use Doctrine\DBAL\DriverManager;
+
+class PoolingShardConnectionTest extends \PHPUnit_Framework_TestCase
+{
+    public function testConnect()
+    {
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'global' => array('memory' => true),
+            'shards' => array(
+                array('id' => 1, 'memory' => true),
+                array('id' => 2, 'memory' => true),
+            ),
+            'shardChoser' => 'Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser',
+        ));
+
+        $this->assertFalse($conn->isConnected(0));
+        $conn->connect(0);
+        $this->assertEquals(1, $conn->fetchColumn('SELECT 1'));
+        $this->assertTrue($conn->isConnected(0));
+
+        $this->assertFalse($conn->isConnected(1));
+        $conn->connect(1);
+        $this->assertEquals(1, $conn->fetchColumn('SELECT 1'));
+        $this->assertTrue($conn->isConnected(1));
+
+        $this->assertFalse($conn->isConnected(2));
+        $conn->connect(2);
+        $this->assertEquals(1, $conn->fetchColumn('SELECT 1'));
+        $this->assertTrue($conn->isConnected(2));
+
+        $conn->close();
+        $this->assertFalse($conn->isConnected(0));
+        $this->assertFalse($conn->isConnected(1));
+        $this->assertFalse($conn->isConnected(2));
+    }
+
+    public function testNoGlobalServerException()
+    {
+        $this->setExpectedException('InvalidArgumentException', "Connection Parameters require 'global' and 'shards' configurations.");
+
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'shards' => array(
+                array('id' => 1, 'memory' => true),
+                array('id' => 2, 'memory' => true),
+            ),
+            'shardChoser' => 'Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser',
+        ));
+    }
+
+    public function testNoShardsServersExecption()
+    {
+        $this->setExpectedException('InvalidArgumentException', "Connection Parameters require 'global' and 'shards' configurations.");
+
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'global' => array('memory' => true),
+            'shardChoser' => 'Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser',
+        ));
+    }
+
+    public function testNoShardsChoserExecption()
+    {
+        $this->setExpectedException('InvalidArgumentException', "Missing Shard Choser configuration 'shardChoser'");
+
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'global' => array('memory' => true),
+            'shards' => array(
+                array('id' => 1, 'memory' => true),
+                array('id' => 2, 'memory' => true),
+            ),
+        ));
+    }
+
+    public function testShardChoserWrongInstance()
+    {
+        $this->setExpectedException('InvalidArgumentException', "The 'shardChoser' configuration is not a valid instance of Doctrine\DBAL\Sharding\ShardChoser\ShardChoser");
+
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'global' => array('memory' => true),
+            'shards' => array(
+                array('id' => 1, 'memory' => true),
+                array('id' => 2, 'memory' => true),
+            ),
+            'shardChoser' => new \stdClass,
+        ));
+    }
+
+    public function testShardNonNumericId()
+    {
+        $this->setExpectedException('InvalidArgumentException', "Shard Id has to be a non-negative number.");
+
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'global' => array('memory' => true),
+            'shards' => array(
+                array('id' => 'foo', 'memory' => true),
+            ),
+            'shardChoser' => 'Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser',
+        ));
+    }
+
+    public function testShardMissingId()
+    {
+        $this->setExpectedException('InvalidArgumentException', "Missing 'id' for one configured shard. Please specificy a unique shard-id.");
+
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'global' => array('memory' => true),
+            'shards' => array(
+                array('memory' => true),
+            ),
+            'shardChoser' => 'Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser',
+        ));
+    }
+
+    public function testDuplicateShardId()
+    {
+        $this->setExpectedException('InvalidArgumentException', "Shard 1 is duplicated in the configuration.");
+
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'global' => array('memory' => true),
+            'shards' => array(
+                array('id' => 1, 'memory' => true),
+                array('id' => 1, 'memory' => true),
+            ),
+            'shardChoser' => 'Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser',
+        ));
+    }
+
+    public function testSwitchShardWithOpenTransactionException()
+    {
+        $conn = DriverManager::getConnection(array(
+            'wrapperClass' => 'Doctrine\DBAL\Sharding\PoolingShardConnection',
+            'driver' => 'pdo_sqlite',
+            'global' => array('memory' => true),
+            'shards' => array(
+                array('id' => 1, 'memory' => true),
+            ),
+            'shardChoser' => 'Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser',
+        ));
+
+        $conn->beginTransaction();
+
+        $this->setExpectedException('Doctrine\DBAL\Sharding\ShardingException', 'Cannot switch shard when transaction is active.');
+        $conn->connect(1);
+    }
+}
+

tests/Doctrine/Tests/DBAL/Sharding/PoolingShardManagerTest.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+namespace Doctrine\Tests\DBAL\Sharding;
+
+use Doctrine\DBAL\Sharding\PoolingShardManager;
+
+class PoolingShardManagerTest extends \PHPUnit_Framework_TestCase
+{
+    private function createConnectionMock()
+    {
+        return $this->getMock('Doctrine\DBAL\Sharding\PoolingShardConnection', array('connect', 'getParams', 'fetchAll'), array(), '', false);
+    }
+
+    private function createPassthroughShardChoser()
+    {
+        $mock = $this->getMock('Doctrine\DBAL\Sharding\ShardChoser\ShardChoser');
+        $mock->expects($this->any())
+             ->method('pickShard')
+             ->will($this->returnCallback(function($value) { return $value; }));
+        return $mock;
+    }
+
+    public function testSelectGlobal()
+    {
+        $conn = $this->createConnectionMock();
+        $conn->expects($this->once())->method('connect')->with($this->equalTo(0));
+
+        $shardManager = new PoolingShardManager($conn, $this->createPassthroughShardChoser());
+        $shardManager->selectGlobal();
+
+        $this->assertNull($shardManager->getCurrentDistributionValue());
+    }
+
+    public function testSelectShard()
+    {
+        $shardId = 10;
+        $conn = $this->createConnectionMock();
+        $conn->expects($this->once())->method('connect')->with($this->equalTo($shardId));
+
+        $shardManager = new PoolingShardManager($conn, $this->createPassthroughShardChoser());
+        $shardManager->selectShard($shardId);
+
+        $this->assertEquals($shardId, $shardManager->getCurrentDistributionValue());
+    }
+
+    public function testGetShards()
+    {
+        $conn = $this->createConnectionMock();
+        $conn->expects($this->once())->method('getParams')->will($this->returnValue(
+            array('shards' => array( array('id' => 1), array('id' => 2) ))
+        ));
+
+        $shardManager = new PoolingShardManager($conn, $this->createPassthroughShardChoser());
+        $shards = $shardManager->getShards();
+
+        $this->assertEquals(array(array('id' => 1), array('id' => 2)), $shards);
+    }
+
+    public function testQueryAll()
+    {
+        $sql = "SELECT * FROM table";
+        $params = array(1);
+        $types = array(1);
+
+        $conn = $this->createConnectionMock();
+        $conn->expects($this->at(0))->method('getParams')->will($this->returnValue(
+            array('shards' => array( array('id' => 1), array('id' => 2) ))
+        ));
+        $conn->expects($this->at(1))->method('connect')->with($this->equalTo(1));
+        $conn->expects($this->at(2))
+             ->method('fetchAll')
+             ->with($this->equalTo($sql), $this->equalTo($params), $this->equalTo($types))
+             ->will($this->returnValue(array( array('id' => 1) ) ));
+        $conn->expects($this->at(3))->method('connect')->with($this->equalTo(2));
+        $conn->expects($this->at(4))
+             ->method('fetchAll')
+             ->with($this->equalTo($sql), $this->equalTo($params), $this->equalTo($types))
+             ->will($this->returnValue(array( array('id' => 2) ) ));
+
+        $shardManager = new PoolingShardManager($conn, $this->createPassthroughShardChoser());
+        $result = $shardManager->queryAll($sql, $params, $types);
+
+        $this->assertEquals(array(array('id' => 1), array('id' => 2)), $result);
+    }
+}
+

tests/Doctrine/Tests/DBAL/Sharding/SQLAzure/AbstractTestCase.php

+<?php
+
+namespace Doctrine\Tests\DBAL\Sharding\SQLAzure;
+
+use Doctrine\DBAL\DriverManager;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Sharding\SQLAzure\SQLAzureShardManager;
+
+abstract class AbstractTestCase extends \PHPUnit_Framework_TestCase
+{
+    protected $conn;
+    protected $sm;
+
+    public function setUp()
+    {
+        if (!isset($GLOBALS['db_type']) || strpos($GLOBALS['db_type'], "sqlsrv") === false) {
+            $this->markTestSkipped('No driver or sqlserver driver specified.');
+        }
+
+        $params = array(
+            'driver' => $GLOBALS['db_type'],
+            'dbname' => $GLOBALS['db_name'],
+            'user' => $GLOBALS['db_username'],
+            'password' => $GLOBALS['db_password'],
+            'host' => $GLOBALS['db_host'],
+            'sharding' => array(
+                'federationName' => 'Orders_Federation',
+                'distributionKey' => 'CustID',
+                'distributionType' => 'integer',
+                'filteringEnabled' => false,
+            ),
+            'driverOptions' => array('MultipleActiveResultSets' => false)
+        );
+        $this->conn = DriverManager::getConnection($params);
+        // assume database is created and schema is:
+        // Global products table
+        // Customers, Orders, OrderItems federation tables.
+        // See http://cloud.dzone.com/articles/using-sql-azure-federations
+        $this->sm = new SQLAzureShardManager($this->conn);
+    }
+
+    public function createShopSchema()
+    {
+        $schema = new Schema();
+
+        $products = $schema->createTable('Products');
+        $products->addColumn('ProductID', 'integer');
+        $products->addColumn('SupplierID', 'integer');
+        $products->addColumn('ProductName', 'string');
+        $products->addColumn('Price', 'decimal', array('scale' => 2, 'precision' => 12));
+        $products->setPrimaryKey(array('ProductID'));
+        $products->addOption('azure.federated', true);
+
+        $customers = $schema->createTable('Customers');
+        $customers->addColumn('CustomerID', 'integer');
+        $customers->addColumn('CompanyName', 'string');
+        $customers->addColumn('FirstName', 'string');
+        $customers->addColumn('LastName', 'string');
+        $customers->setPrimaryKey(array('CustomerID'));
+        $customers->addOption('azure.federated', true);
+        $customers->addOption('azure.federatedOnColumnName', 'CustomerID');
+
+        $orders = $schema->createTable('Orders');
+        $orders->addColumn('CustomerID', 'integer');
+        $orders->addColumn('OrderID', 'integer');
+        $orders->addColumn('OrderDate', 'datetime');
+        $orders->setPrimaryKey(array('CustomerID', 'OrderID'));
+        $orders->addOption('azure.federated', true);
+        $orders->addOption('azure.federatedOnColumnName', 'CustomerID');
+
+        $orderItems = $schema->createTable('OrderItems');
+        $orderItems->addColumn('CustomerID', 'integer');
+        $orderItems->addColumn('OrderID', 'integer');
+        $orderItems->addColumn('ProductID', 'integer');
+        $orderItems->addColumn('Quantity', 'integer');
+        $orderItems->setPrimaryKey(array('CustomerID', 'OrderID', 'ProductID'));
+        $orderItems->addOption('azure.federated', true);
+        $orderItems->addOption('azure.federatedOnColumnName', 'CustomerID');
+
+        return $schema;
+    }
+}

tests/Doctrine/Tests/DBAL/Sharding/SQLAzure/FunctionalTest.php

+<?php
+namespace Doctrine\Tests\DBAL\Sharding\SQLAzure;
+
+use Doctrine\DBAL\Sharding\SQLAzure\SQLAzureSchemaSynchronizer;
+
+class FunctionalTest extends AbstractTestCase
+{
+    public function testSharding()
+    {
+        $schema = $this->createShopSchema();
+
+        $synchronizer = new SQLAzureSchemaSynchronizer($this->conn, $this->sm);
+        $synchronizer->dropAllSchema();
+        $synchronizer->createSchema($schema);
+
+        $this->sm->selectShard(0);
+
+        $this->conn->insert("Products", array(
+            "ProductID" => 1,
+            "SupplierID" => 2,
+            "ProductName" => "Test",
+            "Price" => 10.45
+        ));
+
+        $this->conn->insert("Customers", array(
+            "CustomerID" => 1,
+            "CompanyName" => "Foo",
+            "FirstName" => "Benjamin",
+            "LastName" => "E.",
+        ));
+
+        $query = "SELECT * FROM Products";
+        $data = $this->conn->fetchAll($query);
+        $this->assertTrue(count($data) > 0);
+
+        $query = "SELECT * FROM Customers";
+        $data = $this->conn->fetchAll($query);
+        $this->assertTrue(count($data) > 0);
+
+        $data = $this->sm->queryAll("SELECT * FROM Customers");
+        $this->assertTrue(count($data) > 0);
+    }
+}
+

tests/Doctrine/Tests/DBAL/Sharding/SQLAzure/MultiTenantVisitorTest.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\Tests\DBAL\Sharding\SQLAzure;
+
+use Doctrine\DBAL\Platforms\SQLAzurePlatform;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Sharding\SQLAzure\Schema\MultiTenantVisitor;
+
+class MultiTenantVisitorTest extends \PHPUnit_Framework_TestCase
+{
+    public function testMultiTenantPrimaryKey()
+    {
+        $platform = new SQLAzurePlatform();
+        $visitor = new MultiTenantVisitor();
+
+        $schema = new Schema();
+        $foo = $schema->createTable('foo');
+        $foo->addColumn('id', 'string');
+        $foo->setPrimaryKey(array('id'));
+        $schema->visit($visitor);
+
+        $this->assertEquals(array('id', 'tenant_id'), $foo->getPrimaryKey()->getColumns());
+        $this->assertTrue($foo->hasColumn('tenant_id'));
+    }
+
+    public function testMultiTenantNonPrimaryKey()
+    {
+        $platform = new SQLAzurePlatform();
+        $visitor = new MultiTenantVisitor();
+
+        $schema = new Schema();
+        $foo = $schema->createTable('foo');
+        $foo->addColumn('id', 'string');
+        $foo->addColumn('created', 'datetime');
+        $foo->setPrimaryKey(array('id'));
+        $foo->addIndex(array('created'), 'idx');
+
+        $foo->getPrimaryKey()->addFlag('nonclustered');
+        $foo->getIndex('idx')->addFlag('clustered');
+
+        $schema->visit($visitor);
+
+        $this->assertEquals(array('id'), $foo->getPrimaryKey()->getColumns());
+        $this->assertTrue($foo->hasColumn('tenant_id'));
+        $this->assertEquals(array('created', 'tenant_id'), $foo->getIndex('idx')->getColumns());
+    }
+}
+

tests/Doctrine/Tests/DBAL/Sharding/SQLAzure/SQLAzureSchemaSynchronizerTest.php

+<?php
+namespace Doctrine\Tests\DBAL\Sharding\SQLAzure;
+
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Sharding\SQLAzure\SQLAzureSchemaSynchronizer;
+
+class SQLAzureSchemaSynchronizerTest extends AbstractTestCase
+{
+    public function testCreateSchema()
+    {
+        $schema = $this->createShopSchema();
+
+        $synchronizer = new SQLAzureSchemaSynchronizer($this->conn, $this->sm);
+        $sql = $synchronizer->getCreateSchema($schema);
+
+        $this->assertEquals(array (
+            "--Create Federation\nCREATE FEDERATION Orders_Federation (CustID INT  RANGE)",
+            "USE FEDERATION Orders_Federation (CustID = 0) WITH RESET, FILTERING = OFF;",
+            "CREATE TABLE Products (ProductID INT NOT NULL, SupplierID INT NOT NULL, ProductName NVARCHAR(255) NOT NULL, Price NUMERIC(12, 2) NOT NULL, PRIMARY KEY (ProductID))",
+            "CREATE TABLE Customers (CustomerID INT NOT NULL, CompanyName NVARCHAR(255) NOT NULL, FirstName NVARCHAR(255) NOT NULL, LastName NVARCHAR(255) NOT NULL, PRIMARY KEY (CustomerID))",
+            "CREATE TABLE Orders (CustomerID INT NOT NULL, OrderID INT NOT NULL, OrderDate DATETIME2(6) NOT NULL, PRIMARY KEY (CustomerID, OrderID))",
+            "CREATE TABLE OrderItems (CustomerID INT NOT NULL, OrderID INT NOT NULL, ProductID INT NOT NULL, Quantity INT NOT NULL, PRIMARY KEY (CustomerID, OrderID, ProductID))",
+        ), $sql);
+    }
+
+    public function testUpdateSchema()
+    {
+        $schema = $this->createShopSchema();
+
+        $synchronizer = new SQLAzureSchemaSynchronizer($this->conn, $this->sm);
+        $synchronizer->dropAllSchema();
+
+        $sql = $synchronizer->getUpdateSchema($schema);
+
+        $this->assertEquals(array(), $sql);
+    }
+
+    public function testDropSchema()
+    {
+        $schema = $this->createShopSchema();
+
+        $synchronizer = new SQLAzureSchemaSynchronizer($this->conn, $this->sm);
+        $synchronizer->dropAllSchema();
+        $synchronizer->createSchema($schema);
+        $sql = $synchronizer->getDropSchema($schema);
+
+        $this->assertEQuals(5, count($sql));
+    }
+}
+

tests/Doctrine/Tests/DBAL/Sharding/SQLAzure/SQLAzureShardManagerTest.php

+<?php
+
+namespace Doctrine\Tests\DBAL\Sharding\SQLAzure;
+
+use Doctrine\DBAL\Sharding\SQLAzure\SQLAzureShardManager;
+
+class SQLAzureShardManagerTest extends \PHPUnit_Framework_TestCase
+{
+    public function testNoFederationName()
+    {
+        $this->setExpectedException('Doctrine\DBAL\Sharding\ShardingException', 'SQLAzure requires a federation name to be set during sharding configuration.');
+
+        $conn = $this->createConnection(array('sharding' => array('distributionKey' => 'abc', 'distributionType' => 'integer')));
+        $sm = new SQLAzureShardManager($conn);
+    }
+
+    public function testNoDistributionKey()
+    {
+        $this->setExpectedException('Doctrine\DBAL\Sharding\ShardingException', 'SQLAzure requires a distribution key to be set during sharding configuration.');
+
+        $conn = $this->createConnection(array('sharding' => array('federationName' => 'abc', 'distributionType' => 'integer')));
+        $sm = new SQLAzureShardManager($conn);
+    }
+
+    public function testNoDistributionType()
+    {
+        $this->setExpectedException('Doctrine\DBAL\Sharding\ShardingException');
+
+        $conn = $this->createConnection(array('sharding' => array('federationName' => 'abc', 'distributionKey' => 'foo')));
+        $sm = new SQLAzureShardManager($conn);
+    }
+
+    public function testGetDefaultDistributionValue()
+    {
+        $conn = $this->createConnection(array('sharding' => array('federationName' => 'abc', 'distributionKey' => 'foo', 'distributionType' => 'integer')));
+
+        $sm = new SQLAzureShardManager($conn);
+        $this->assertNull($sm->getCurrentDistributionValue());
+    }
+
+    public function testSelectGlobalTransactionActive()
+    {
+        $conn = $this->createConnection(array('sharding' => array('federationName' => 'abc', 'distributionKey' => 'foo', 'distributionType' => 'integer')));
+        $conn->expects($this->at(1))->method('isTransactionActive')->will($this->returnValue(true));
+
+        $this->setExpectedException('Doctrine\DBAL\Sharding\ShardingException', 'Cannot switch shard during an active transaction.');
+
+        $sm = new SQLAzureShardManager($conn);
+        $sm->selectGlobal();
+    }
+
+    public function testSelectGlobal()
+    {
+        $conn = $this->createConnection(array('sharding' => array('federationName' => 'abc', 'distributionKey' => 'foo', 'distributionType' => 'integer')));
+        $conn->expects($this->at(1))->method('isTransactionActive')->will($this->returnValue(false));
+        $conn->expects($this->at(2))->method('exec')->with($this->equalTo('USE FEDERATION ROOT WITH RESET'));
+
+        $sm = new SQLAzureShardManager($conn);
+        $sm->selectGlobal();
+    }
+
+    public function testSelectShard()
+    {
+        $conn = $this->createConnection(array('sharding' => array('federationName' => 'abc', 'distributionKey' => 'foo', 'distributionType' => 'integer')));
+        $conn->expects($this->at(1))->method('isTransactionActive')->will($this->returnValue(true));
+
+        $this->setExpectedException('Doctrine\DBAL\Sharding\ShardingException', 'Cannot switch shard during an active transaction.');
+
+        $sm = new SQLAzureShardManager($conn);
+        $sm->selectShard(1234);
+
+        $this->assertEquals(1234, $sm->getCurrentDistributionValue());
+    }
+
+    public function testSelectShardNoDistriubtionValue()
+    {
+        $conn = $this->createConnection(array('sharding' => array('federationName' => 'abc', 'distributionKey' => 'foo', 'distributionType' => 'integer')));
+        $conn->expects($this->at(1))->method('isTransactionActive')->will($this->returnValue(false));
+
+        $this->setExpectedException('Doctrine\DBAL\Sharding\ShardingException', 'You have to specify a string or integer as shard distribution value.');
+
+        $sm = new SQLAzureShardManager($conn);
+        $sm->selectShard(null);
+    }
+
+    private function createConnection(array $params)
+    {
+        $conn = $this->getMock('Doctrine\DBAL\Connection', array('getParams', 'exec', 'isTransactionActive'), array(), '', false);
+        $conn->expects($this->at(0))->method('getParams')->will($this->returnValue($params));
+        return $conn;
+    }
+}
+

tests/Doctrine/Tests/DBAL/Sharding/ShardChoser/MultiTenantShardChoserTest.php

+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the LGPL. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\Tests\DBAL\Sharding\ShardChoser;
+
+use Doctrine\DBAL\Sharding\ShardChoser\MultiTenantShardChoser;
+
+class MultiTenantShardChoserTest extends \PHPUnit_Framework_TestCase
+{
+    public function testPickShard()
+    {
+        $choser = new MultiTenantShardChoser();
+        $conn = $this->createConnectionMock();
+
+        $this->assertEquals(1, $choser->pickShard(1, $conn));
+        $this->assertEquals(2, $choser->pickShard(2, $conn));
+    }
+
+    private function createConnectionMock()
+    {
+        return $this->getMock('Doctrine\DBAL\Sharding\PoolingShardConnection', array('connect', 'getParams', 'fetchAll'), array(), '', false);
+    }
+}
+