De-normalize tags used to retrieve builds queue

app/models/ci/pending_build.rb

@@ -12,51 +12,55 @@ module Ci
     scope :queued_before, ->(time) { where(arel_table[:created_at].lt(time)) }
     scope :with_instance_runners, -> { where(instance_runners_enabled: true) }
 
-    def self.upsert_from_build!(build)
-      entry = self.new(args_from_build(build))
+    class << self
+      def upsert_from_build!(build)
+        entry = self.new(args_from_build(build))
 
-      entry.validate!
+        entry.validate!
 
-      self.upsert(entry.attributes.compact, returning: %w[build_id], unique_by: :build_id)
-    end
+        self.upsert(entry.attributes.compact, returning: %w[build_id], unique_by: :build_id)
+      end
+
+      private
+
+      def args_from_build(build)
+        args = {
+          build: build,
+          project: build.project,
+          protected: build.protected?,
+          namespace: build.project.namespace
+        }
+
+        if Feature.enabled?(:ci_pending_builds_maintain_tags_data, type: :development, default_enabled: :yaml)
+          args.store(:tag_ids, build.tags_ids)
+        end
+
+        if Feature.enabled?(:ci_pending_builds_maintain_shared_runners_data, type: :development, default_enabled: :yaml)
+          args.store(:instance_runners_enabled, shareable?(build))
+        end
 
-    def self.args_from_build(build)
-      args = {
-        build: build,
-        project: build.project,
-        protected: build.protected?,
-        namespace: build.project.namespace
-      }
-
-      if Feature.enabled?(:ci_pending_builds_maintain_shared_runners_data, type: :development, default_enabled: :yaml)
-        args.merge(instance_runners_enabled: shareable?(build))
-      else
         args
       end
-    end
-    private_class_method :args_from_build
 
-    def self.shareable?(build)
-      shared_runner_enabled?(build) &&
-        builds_access_level?(build) &&
-        project_not_removed?(build)
-    end
-    private_class_method :shareable?
+      def shareable?(build)
+        shared_runner_enabled?(build) &&
+          builds_access_level?(build) &&
+          project_not_removed?(build)
+      end
 
-    def self.shared_runner_enabled?(build)
-      build.project.shared_runners.exists?
-    end
-    private_class_method :shared_runner_enabled?
+      def shared_runner_enabled?(build)
+        build.project.shared_runners.exists?
+      end
 
-    def self.project_not_removed?(build)
-      !build.project.pending_delete?
-    end
-    private_class_method :project_not_removed?
+      def project_not_removed?(build)
+        !build.project.pending_delete?
+      end
 
-    def self.builds_access_level?(build)
-      build.project.project_feature.builds_access_level.nil? || build.project.project_feature.builds_access_level > 0
+      def builds_access_level?(build)
+        build.project.project_feature.builds_access_level.nil? ||
+          build.project.project_feature.builds_access_level > 0
+      end
     end
-    private_class_method :builds_access_level?
   end
 end
 

app/models/concerns/taggable_queries.rb

@@ -3,6 +3,10 @@
 module TaggableQueries
   extend ActiveSupport::Concern
 
+  MAX_TAGS_IDS = 50
+
+  TooManyTagsError = Class.new(StandardError)
+
   class_methods do
     # context is a name `acts_as_taggable context`
     def arel_tag_names_array(context = :tags)
@@ -34,4 +38,10 @@ module TaggableQueries
       where("EXISTS (?)", matcher)
     end
   end
+
+  def tags_ids
+    tags.limit(MAX_TAGS_IDS).order('id ASC').pluck(:id).tap do |ids|
+      raise TooManyTagsError if ids.size >= MAX_TAGS_IDS
+    end
+  end
 end

app/services/ci/queue/pending_builds_strategy.rb

@@ -17,11 +17,19 @@ module Ci
       end
 
       def builds_matching_tag_ids(relation, ids)
-        relation.merge(CommitStatus.matches_tag_ids(ids, table: 'ci_pending_builds', column: 'build_id'))
+        if ::Feature.enabled?(:ci_queueing_denormalize_tags_information, runner, default_enabled: :yaml)
+          relation.where('tag_ids <@ ARRAY[?]::int[]', runner.tags_ids)
+        else
+          relation.merge(CommitStatus.matches_tag_ids(ids, table: 'ci_pending_builds', column: 'build_id'))
+        end
       end
 
       def builds_with_any_tags(relation)
-        relation.merge(CommitStatus.with_any_tags(table: 'ci_pending_builds', column: 'build_id'))
+        if ::Feature.enabled?(:ci_queueing_denormalize_tags_information, runner, default_enabled: :yaml)
+          relation.where('cardinality(tag_ids) > 0')
+        else
+          relation.merge(CommitStatus.with_any_tags(table: 'ci_pending_builds', column: 'build_id'))
+        end
       end
 
       def order(relation)

config/feature_flags/development/ci_pending_builds_maintain_tags_data.yml

+---
+name: ci_pending_builds_maintain_tags_data
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/65648
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/338363
+milestone: '14.2'
+type: development
+group: group::pipeline execution
+default_enabled: false

config/feature_flags/development/ci_queueing_denormalize_tags_information.yml

+---
+name: ci_queueing_denormalize_tags_information
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/65648
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/338366
+milestone: '14.1'
+type: development
+group: group::pipeline execution
+default_enabled: false

db/migrate/20210707113056_add_tags_array_to_ci_pending_builds.rb

+# frozen_string_literal: true
+
+class AddTagsArrayToCiPendingBuilds < ActiveRecord::Migration[6.1]
+  include Gitlab::Database::MigrationHelpers
+
+  def up
+    with_lock_retries do
+      add_column :ci_pending_builds, :tag_ids, :integer, array: true, default: []
+    end
+  end
+
+  def down
+    with_lock_retries do
+      remove_column :ci_pending_builds, :tag_ids
+    end
+  end
+end

db/migrate/20210818185548_add_tag_ids_index_to_ci_pending_build.rb

+# frozen_string_literal: true
+
+class AddTagIdsIndexToCiPendingBuild < ActiveRecord::Migration[6.1]
+  include Gitlab::Database::MigrationHelpers
+
+  disable_ddl_transaction!
+
+  INDEX_NAME = 'index_ci_pending_builds_on_tag_ids'
+
+  def up
+    add_concurrent_index(:ci_pending_builds, :tag_ids, name: INDEX_NAME, where: 'cardinality(tag_ids) > 0')
+  end
+
+  def down
+    remove_concurrent_index_by_name(:ci_pending_builds, name: INDEX_NAME)
+  end
+end

db/schema_migrations/20210707113056

+837b9a56114c63064379cf276a3c7e2bbe845af9022a542c4fcec94a25062017
\ No newline at end of file

db/schema_migrations/20210818185548

+88ca485c8513df96b1f1aec1585c385223dc53889e547db42b509b0cd1bea9b7
\ No newline at end of file

db/structure.sql

@@ -10903,7 +10903,8 @@ CREATE TABLE ci_pending_builds (
     protected boolean DEFAULT false NOT NULL,
     instance_runners_enabled boolean DEFAULT false NOT NULL,
     namespace_id bigint,
-    minutes_exceeded boolean DEFAULT false NOT NULL
+    minutes_exceeded boolean DEFAULT false NOT NULL,
+    tag_ids integer[] DEFAULT '{}'::integer[]
 );
 
 CREATE SEQUENCE ci_pending_builds_id_seq
@@ -23439,6 +23440,8 @@ CREATE INDEX index_ci_pending_builds_on_namespace_id ON ci_pending_builds USING
 
 CREATE INDEX index_ci_pending_builds_on_project_id ON ci_pending_builds USING btree (project_id);
 
+CREATE INDEX index_ci_pending_builds_on_tag_ids ON ci_pending_builds USING btree (tag_ids) WHERE (cardinality(tag_ids) > 0);
+
 CREATE INDEX index_ci_pipeline_artifacts_failed_verification ON ci_pipeline_artifacts USING btree (verification_retry_at NULLS FIRST) WHERE (verification_state = 3);
 
 CREATE INDEX index_ci_pipeline_artifacts_needs_verification ON ci_pipeline_artifacts USING btree (verification_state) WHERE ((verification_state = 0) OR (verification_state = 3));

spec/models/ci/pending_build_spec.rb

@@ -113,5 +113,31 @@ RSpec.describe Ci::PendingBuild do
         end
       end
     end
+
+    context 'when build has tags' do
+      let!(:build) { create(:ci_build, :tags) }
+
+      subject(:ci_pending_build) { described_class.last }
+
+      context 'when ci_pending_builds_maintain_tags_data is enabled' do
+        it 'sets tag_ids' do
+          described_class.upsert_from_build!(build)
+
+          expect(ci_pending_build.tag_ids).to eq(build.tags_ids)
+        end
+      end
+
+      context 'when ci_pending_builds_maintain_tags_data is disabled' do
+        before do
+          stub_feature_flags(ci_pending_builds_maintain_tags_data: false)
+        end
+
+        it 'does not set tag_ids' do
+          described_class.upsert_from_build!(build)
+
+          expect(ci_pending_build.tag_ids).to be_empty
+        end
+      end
+    end
   end
 end

spec/services/ci/register_job_service_spec.rb

@@ -739,6 +739,22 @@ module Ci
 
           include_examples 'handles runner assignment'
         end
+
+        context 'with ci_queueing_denormalize_tags_information enabled' do
+          before do
+            stub_feature_flags(ci_queueing_denormalize_tags_information: true)
+          end
+
+          include_examples 'handles runner assignment'
+        end
+
+        context 'with ci_queueing_denormalize_tags_information disabled' do
+          before do
+            stub_feature_flags(ci_queueing_denormalize_tags_information: false)
+          end
+
+          include_examples 'handles runner assignment'
+        end
       end
 
       context 'when not using pending builds table' do