SQLAzureShardManager.php 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214
  1. <?php
  2. namespace Doctrine\DBAL\Sharding\SQLAzure;
  3. use Doctrine\DBAL\Connection;
  4. use Doctrine\DBAL\Sharding\ShardingException;
  5. use Doctrine\DBAL\Sharding\ShardManager;
  6. use Doctrine\DBAL\Types\Type;
  7. use RuntimeException;
  8. use function is_bool;
  9. use function is_scalar;
  10. use function sprintf;
  11. /**
  12. * Sharding using the SQL Azure Federations support.
  13. */
  14. class SQLAzureShardManager implements ShardManager
  15. {
  16. /** @var string */
  17. private $federationName;
  18. /** @var bool */
  19. private $filteringEnabled;
  20. /** @var string */
  21. private $distributionKey;
  22. /** @var string */
  23. private $distributionType;
  24. /** @var Connection */
  25. private $conn;
  26. /** @var string|null */
  27. private $currentDistributionValue;
  28. /**
  29. * @throws ShardingException
  30. */
  31. public function __construct(Connection $conn)
  32. {
  33. $this->conn = $conn;
  34. $params = $conn->getParams();
  35. if (! isset($params['sharding']['federationName'])) {
  36. throw ShardingException::missingDefaultFederationName();
  37. }
  38. if (! isset($params['sharding']['distributionKey'])) {
  39. throw ShardingException::missingDefaultDistributionKey();
  40. }
  41. if (! isset($params['sharding']['distributionType'])) {
  42. throw ShardingException::missingDistributionType();
  43. }
  44. $this->federationName = $params['sharding']['federationName'];
  45. $this->distributionKey = $params['sharding']['distributionKey'];
  46. $this->distributionType = $params['sharding']['distributionType'];
  47. $this->filteringEnabled = (bool) ($params['sharding']['filteringEnabled'] ?? false);
  48. }
  49. /**
  50. * Gets the name of the federation.
  51. *
  52. * @return string
  53. */
  54. public function getFederationName()
  55. {
  56. return $this->federationName;
  57. }
  58. /**
  59. * Gets the distribution key.
  60. *
  61. * @return string
  62. */
  63. public function getDistributionKey()
  64. {
  65. return $this->distributionKey;
  66. }
  67. /**
  68. * Gets the Doctrine Type name used for the distribution.
  69. *
  70. * @return string
  71. */
  72. public function getDistributionType()
  73. {
  74. return $this->distributionType;
  75. }
  76. /**
  77. * Sets Enabled/Disable filtering on the fly.
  78. *
  79. * @param bool $flag
  80. *
  81. * @return void
  82. */
  83. public function setFilteringEnabled($flag)
  84. {
  85. $this->filteringEnabled = (bool) $flag;
  86. }
  87. /**
  88. * {@inheritDoc}
  89. */
  90. public function selectGlobal()
  91. {
  92. if ($this->conn->isTransactionActive()) {
  93. throw ShardingException::activeTransaction();
  94. }
  95. $sql = 'USE FEDERATION ROOT WITH RESET';
  96. $this->conn->exec($sql);
  97. $this->currentDistributionValue = null;
  98. }
  99. /**
  100. * {@inheritDoc}
  101. */
  102. public function selectShard($distributionValue)
  103. {
  104. if ($this->conn->isTransactionActive()) {
  105. throw ShardingException::activeTransaction();
  106. }
  107. if ($distributionValue === null || is_bool($distributionValue) || ! is_scalar($distributionValue)) {
  108. throw ShardingException::noShardDistributionValue();
  109. }
  110. $platform = $this->conn->getDatabasePlatform();
  111. $sql = sprintf(
  112. 'USE FEDERATION %s (%s = %s) WITH RESET, FILTERING = %s;',
  113. $platform->quoteIdentifier($this->federationName),
  114. $platform->quoteIdentifier($this->distributionKey),
  115. $this->conn->quote($distributionValue),
  116. ($this->filteringEnabled ? 'ON' : 'OFF')
  117. );
  118. $this->conn->exec($sql);
  119. $this->currentDistributionValue = $distributionValue;
  120. }
  121. /**
  122. * {@inheritDoc}
  123. */
  124. public function getCurrentDistributionValue()
  125. {
  126. return $this->currentDistributionValue;
  127. }
  128. /**
  129. * {@inheritDoc}
  130. */
  131. public function getShards()
  132. {
  133. $sql = 'SELECT member_id as id,
  134. distribution_name as distribution_key,
  135. CAST(range_low AS CHAR) AS rangeLow,
  136. CAST(range_high AS CHAR) AS rangeHigh
  137. FROM sys.federation_member_distributions d
  138. INNER JOIN sys.federations f ON f.federation_id = d.federation_id
  139. WHERE f.name = ' . $this->conn->quote($this->federationName);
  140. return $this->conn->fetchAll($sql);
  141. }
  142. /**
  143. * {@inheritDoc}
  144. */
  145. public function queryAll($sql, array $params = [], array $types = [])
  146. {
  147. $shards = $this->getShards();
  148. if (! $shards) {
  149. throw new RuntimeException('No shards found for ' . $this->federationName);
  150. }
  151. $result = [];
  152. $oldDistribution = $this->getCurrentDistributionValue();
  153. foreach ($shards as $shard) {
  154. $this->selectShard($shard['rangeLow']);
  155. foreach ($this->conn->fetchAll($sql, $params, $types) as $row) {
  156. $result[] = $row;
  157. }
  158. }
  159. if ($oldDistribution === null) {
  160. $this->selectGlobal();
  161. } else {
  162. $this->selectShard($oldDistribution);
  163. }
  164. return $result;
  165. }
  166. /**
  167. * Splits Federation at a given distribution value.
  168. *
  169. * @param mixed $splitDistributionValue
  170. *
  171. * @return void
  172. */
  173. public function splitFederation($splitDistributionValue)
  174. {
  175. $type = Type::getType($this->distributionType);
  176. $sql = 'ALTER FEDERATION ' . $this->getFederationName() . ' ' .
  177. 'SPLIT AT (' . $this->getDistributionKey() . ' = ' .
  178. $this->conn->quote($splitDistributionValue, $type->getBindingType()) . ')';
  179. $this->conn->exec($sql);
  180. }
  181. }