1 files changed, 34 insertions, 7 deletions

diff --git a/app/workers/database/batched_background_migration/single_database_worker.rb b/app/workers/database/batched_background_migration/single_database_worker.rb
index 0c7c51d5c0a..e772216e557 100644
--- a/app/workers/database/batched_background_migration/single_database_worker.rb
+++ b/app/workers/database/batched_background_migration/single_database_worker.rb
@@ -39,7 +39,7 @@ module Database
         unless base_model
           Sidekiq.logger.info(
             class: self.class.name,
-            database: self.class.tracking_database,
+            database: tracking_database,
             message: 'skipping migration execution for unconfigured database')
 
           return
@@ -48,34 +48,61 @@ module Database
         if shares_db_config?
           Sidekiq.logger.info(
             class: self.class.name,
-            database: self.class.tracking_database,
+            database: tracking_database,
             message: 'skipping migration execution for database that shares database configuration with another database')
 
           return
         end
 
         Gitlab::Database::SharedModel.using_connection(base_model.connection) do
-          break unless self.class.enabled? && active_migration
+          break unless self.class.enabled?
 
-          with_exclusive_lease(active_migration.interval) do
-            run_active_migration
+          if parallel_execution_enabled?
+            migrations = Gitlab::Database::BackgroundMigration::BatchedMigration
+              .active_migrations_distinct_on_table(connection: base_model.connection, limit: max_running_migrations).to_a
+
+            queue_migrations_for_execution(migrations) if migrations.any?
+          else
+            break unless active_migration
+
+            with_exclusive_lease(active_migration.interval) do
+              run_active_migration
+            end
           end
         end
       end
 
       private
 
+      def parallel_execution_enabled?
+        Feature.enabled?(:batched_migrations_parallel_execution)
+      end
+
+      def max_running_migrations
+        execution_worker_class.max_running_jobs
+      end
+
       def active_migration
         @active_migration ||= Gitlab::Database::BackgroundMigration::BatchedMigration.active_migration(connection: base_model.connection)
       end
 
       def run_active_migration
-        Database::BatchedBackgroundMigration::ExecutionWorker.new.perform(self.class.tracking_database, active_migration.id)
+        execution_worker_class.new.perform_work(tracking_database, active_migration.id)
+      end
+
+      def tracking_database
+        self.class.tracking_database
+      end
+
+      def queue_migrations_for_execution(migrations)
+        jobs_arguments = migrations.map { |migration| [tracking_database.to_s, migration.id] }
+
+        execution_worker_class.perform_with_capacity(jobs_arguments)
       end
 
       def base_model
         strong_memoize(:base_model) do
-          Gitlab::Database.database_base_models[self.class.tracking_database]
+          Gitlab::Database.database_base_models[tracking_database]
         end
       end