SQLAzureFederationsSynchronizer.php 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270
  1. <?php
  2. namespace Doctrine\DBAL\Sharding\SQLAzure;
  3. use Closure;
  4. use Doctrine\DBAL\Connection;
  5. use Doctrine\DBAL\Schema\Schema;
  6. use Doctrine\DBAL\Schema\Synchronizer\AbstractSchemaSynchronizer;
  7. use Doctrine\DBAL\Schema\Synchronizer\SchemaSynchronizer;
  8. use Doctrine\DBAL\Schema\Synchronizer\SingleDatabaseSynchronizer;
  9. use Doctrine\DBAL\Types\Type;
  10. use RuntimeException;
  11. use function array_merge;
  12. /**
  13. * SQL Azure Schema Synchronizer.
  14. *
  15. * Will iterate over all shards when performing schema operations. This is done
  16. * by partitioning the passed schema into subschemas for the federation and the
  17. * global database and then applying the operations step by step using the
  18. * {@see \Doctrine\DBAL\Schema\Synchronizer\SingleDatabaseSynchronizer}.
  19. */
  20. class SQLAzureFederationsSynchronizer extends AbstractSchemaSynchronizer
  21. {
  22. public const FEDERATION_TABLE_FEDERATED = 'azure.federated';
  23. public const FEDERATION_DISTRIBUTION_NAME = 'azure.federatedOnDistributionName';
  24. /** @var SQLAzureShardManager */
  25. private $shardManager;
  26. /** @var SchemaSynchronizer */
  27. private $synchronizer;
  28. public function __construct(Connection $conn, SQLAzureShardManager $shardManager, ?SchemaSynchronizer $sync = null)
  29. {
  30. parent::__construct($conn);
  31. $this->shardManager = $shardManager;
  32. $this->synchronizer = $sync ?: new SingleDatabaseSynchronizer($conn);
  33. }
  34. /**
  35. * {@inheritdoc}
  36. */
  37. public function getCreateSchema(Schema $createSchema)
  38. {
  39. $sql = [];
  40. [$global, $federation] = $this->partitionSchema($createSchema);
  41. $globalSql = $this->synchronizer->getCreateSchema($global);
  42. if ($globalSql) {
  43. $sql[] = "-- Create Root Federation\n" .
  44. 'USE FEDERATION ROOT WITH RESET;';
  45. $sql = array_merge($sql, $globalSql);
  46. }
  47. $federationSql = $this->synchronizer->getCreateSchema($federation);
  48. if ($federationSql) {
  49. $defaultValue = $this->getFederationTypeDefaultValue();
  50. $sql[] = $this->getCreateFederationStatement();
  51. $sql[] = 'USE FEDERATION ' . $this->shardManager->getFederationName() . ' (' . $this->shardManager->getDistributionKey() . ' = ' . $defaultValue . ') WITH RESET, FILTERING = OFF;';
  52. $sql = array_merge($sql, $federationSql);
  53. }
  54. return $sql;
  55. }
  56. /**
  57. * {@inheritdoc}
  58. */
  59. public function getUpdateSchema(Schema $toSchema, $noDrops = false)
  60. {
  61. return $this->work($toSchema, static function ($synchronizer, $schema) use ($noDrops) {
  62. return $synchronizer->getUpdateSchema($schema, $noDrops);
  63. });
  64. }
  65. /**
  66. * {@inheritdoc}
  67. */
  68. public function getDropSchema(Schema $dropSchema)
  69. {
  70. return $this->work($dropSchema, static function ($synchronizer, $schema) {
  71. return $synchronizer->getDropSchema($schema);
  72. });
  73. }
  74. /**
  75. * {@inheritdoc}
  76. */
  77. public function createSchema(Schema $createSchema)
  78. {
  79. $this->processSql($this->getCreateSchema($createSchema));
  80. }
  81. /**
  82. * {@inheritdoc}
  83. */
  84. public function updateSchema(Schema $toSchema, $noDrops = false)
  85. {
  86. $this->processSql($this->getUpdateSchema($toSchema, $noDrops));
  87. }
  88. /**
  89. * {@inheritdoc}
  90. */
  91. public function dropSchema(Schema $dropSchema)
  92. {
  93. $this->processSqlSafely($this->getDropSchema($dropSchema));
  94. }
  95. /**
  96. * {@inheritdoc}
  97. */
  98. public function getDropAllSchema()
  99. {
  100. $this->shardManager->selectGlobal();
  101. $globalSql = $this->synchronizer->getDropAllSchema();
  102. if ($globalSql) {
  103. $sql[] = "-- Work on Root Federation\nUSE FEDERATION ROOT WITH RESET;";
  104. $sql = array_merge($sql, $globalSql);
  105. }
  106. $shards = $this->shardManager->getShards();
  107. foreach ($shards as $shard) {
  108. $this->shardManager->selectShard($shard['rangeLow']);
  109. $federationSql = $this->synchronizer->getDropAllSchema();
  110. if (! $federationSql) {
  111. continue;
  112. }
  113. $sql[] = '-- Work on Federation ID ' . $shard['id'] . "\n" .
  114. 'USE FEDERATION ' . $this->shardManager->getFederationName() . ' (' . $this->shardManager->getDistributionKey() . ' = ' . $shard['rangeLow'] . ') WITH RESET, FILTERING = OFF;';
  115. $sql = array_merge($sql, $federationSql);
  116. }
  117. $sql[] = 'USE FEDERATION ROOT WITH RESET;';
  118. $sql[] = 'DROP FEDERATION ' . $this->shardManager->getFederationName();
  119. return $sql;
  120. }
  121. /**
  122. * {@inheritdoc}
  123. */
  124. public function dropAllSchema()
  125. {
  126. $this->processSqlSafely($this->getDropAllSchema());
  127. }
  128. /**
  129. * @return Schema[]
  130. */
  131. private function partitionSchema(Schema $schema)
  132. {
  133. return [
  134. $this->extractSchemaFederation($schema, false),
  135. $this->extractSchemaFederation($schema, true),
  136. ];
  137. }
  138. /**
  139. * @param bool $isFederation
  140. *
  141. * @return Schema
  142. *
  143. * @throws RuntimeException
  144. */
  145. private function extractSchemaFederation(Schema $schema, $isFederation)
  146. {
  147. $partitionedSchema = clone $schema;
  148. foreach ($partitionedSchema->getTables() as $table) {
  149. if ($isFederation) {
  150. $table->addOption(self::FEDERATION_DISTRIBUTION_NAME, $this->shardManager->getDistributionKey());
  151. }
  152. if ($table->hasOption(self::FEDERATION_TABLE_FEDERATED) !== $isFederation) {
  153. $partitionedSchema->dropTable($table->getName());
  154. } else {
  155. foreach ($table->getForeignKeys() as $fk) {
  156. $foreignTable = $schema->getTable($fk->getForeignTableName());
  157. if ($foreignTable->hasOption(self::FEDERATION_TABLE_FEDERATED) !== $isFederation) {
  158. throw new RuntimeException('Cannot have foreign key between global/federation.');
  159. }
  160. }
  161. }
  162. }
  163. return $partitionedSchema;
  164. }
  165. /**
  166. * Work on the Global/Federation based on currently existing shards and
  167. * perform the given operation on the underlying schema synchronizer given
  168. * the different partitioned schema instances.
  169. *
  170. * @return string[]
  171. */
  172. private function work(Schema $schema, Closure $operation)
  173. {
  174. [$global, $federation] = $this->partitionSchema($schema);
  175. $sql = [];
  176. $this->shardManager->selectGlobal();
  177. $globalSql = $operation($this->synchronizer, $global);
  178. if ($globalSql) {
  179. $sql[] = "-- Work on Root Federation\nUSE FEDERATION ROOT WITH RESET;";
  180. $sql = array_merge($sql, $globalSql);
  181. }
  182. $shards = $this->shardManager->getShards();
  183. foreach ($shards as $shard) {
  184. $this->shardManager->selectShard($shard['rangeLow']);
  185. $federationSql = $operation($this->synchronizer, $federation);
  186. if (! $federationSql) {
  187. continue;
  188. }
  189. $sql[] = '-- Work on Federation ID ' . $shard['id'] . "\n" .
  190. 'USE FEDERATION ' . $this->shardManager->getFederationName() . ' (' . $this->shardManager->getDistributionKey() . ' = ' . $shard['rangeLow'] . ') WITH RESET, FILTERING = OFF;';
  191. $sql = array_merge($sql, $federationSql);
  192. }
  193. return $sql;
  194. }
  195. /**
  196. * @return string
  197. */
  198. private function getFederationTypeDefaultValue()
  199. {
  200. $federationType = Type::getType($this->shardManager->getDistributionType());
  201. switch ($federationType->getName()) {
  202. case Type::GUID:
  203. $defaultValue = '00000000-0000-0000-0000-000000000000';
  204. break;
  205. case Type::INTEGER:
  206. case Type::SMALLINT:
  207. case Type::BIGINT:
  208. $defaultValue = '0';
  209. break;
  210. default:
  211. $defaultValue = '';
  212. break;
  213. }
  214. return $defaultValue;
  215. }
  216. /**
  217. * @return string
  218. */
  219. private function getCreateFederationStatement()
  220. {
  221. $federationType = Type::getType($this->shardManager->getDistributionType());
  222. $federationTypeSql = $federationType->getSQLDeclaration([], $this->conn->getDatabasePlatform());
  223. return "--Create Federation\n" .
  224. 'CREATE FEDERATION ' . $this->shardManager->getFederationName() . ' (' . $this->shardManager->getDistributionKey() . ' ' . $federationTypeSql . ' RANGE)';
  225. }
  226. }