Add latest changes from gitlab-org/gitlab@14-3-stable-eev14.3.0-rc42

12 files changed, 643 insertions, 21 deletions

diff --git a/lib/gitlab/pagination/cursor_based_keyset.rb b/lib/gitlab/pagination/cursor_based_keyset.rb
new file mode 100644
index 00000000000..f19cdf06d9a
--- /dev/null
+++ b/lib/gitlab/pagination/cursor_based_keyset.rb
@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Pagination
+    module CursorBasedKeyset
+      SUPPORTED_ORDERING = {
+        Group => { name: :asc }
+      }.freeze
+
+      def self.available_for_type?(relation)
+        SUPPORTED_ORDERING.key?(relation.klass)
+      end
+
+      def self.available?(cursor_based_request_context, relation)
+        available_for_type?(relation) &&
+        order_satisfied?(relation, cursor_based_request_context)
+      end
+
+      def self.order_satisfied?(relation, cursor_based_request_context)
+        order_by_from_request = cursor_based_request_context.order_by
+
+        SUPPORTED_ORDERING[relation.klass] == order_by_from_request
+      end
+      private_class_method :order_satisfied?
+    end
+  end
+end
diff --git a/lib/gitlab/pagination/gitaly_keyset_pager.rb b/lib/gitlab/pagination/gitaly_keyset_pager.rb
index b05891066ac..a16bf7a379c 100644
--- a/lib/gitlab/pagination/gitaly_keyset_pager.rb
+++ b/lib/gitlab/pagination/gitaly_keyset_pager.rb
@@ -14,23 +14,39 @@ module Gitlab
       # It is expected that the given finder will respond to `execute` method with `gitaly_pagination: true` option
       # and supports pagination via gitaly.
       def paginate(finder)
-        return paginate_via_gitaly(finder) if keyset_pagination_enabled?
-        return paginate_first_page_via_gitaly(finder) if paginate_first_page?
+        return paginate_via_gitaly(finder) if keyset_pagination_enabled?(finder)
+        return paginate_first_page_via_gitaly(finder) if paginate_first_page?(finder)
 
-        branches = ::Kaminari.paginate_array(finder.execute)
+        records = ::Kaminari.paginate_array(finder.execute)
         Gitlab::Pagination::OffsetPagination
           .new(request_context)
-          .paginate(branches)
+          .paginate(records)
       end
 
       private
 
-      def keyset_pagination_enabled?
-        Feature.enabled?(:branch_list_keyset_pagination, project, default_enabled: :yaml) && params[:pagination] == 'keyset'
+      def keyset_pagination_enabled?(finder)
+        return false unless params[:pagination] == "keyset"
+
+        if finder.is_a?(BranchesFinder)
+          Feature.enabled?(:branch_list_keyset_pagination, project, default_enabled: :yaml)
+        elsif finder.is_a?(::Repositories::TreeFinder)
+          Feature.enabled?(:repository_tree_gitaly_pagination, project, default_enabled: :yaml)
+        else
+          false
+        end
       end
 
-      def paginate_first_page?
-        Feature.enabled?(:branch_list_keyset_pagination, project, default_enabled: :yaml) && (params[:page].blank? || params[:page].to_i == 1)
+      def paginate_first_page?(finder)
+        return false unless params[:page].blank? || params[:page].to_i == 1
+
+        if finder.is_a?(BranchesFinder)
+          Feature.enabled?(:branch_list_keyset_pagination, project, default_enabled: :yaml)
+        elsif finder.is_a?(::Repositories::TreeFinder)
+          Feature.enabled?(:repository_tree_gitaly_pagination, project, default_enabled: :yaml)
+        else
+          false
+        end
       end
 
       def paginate_via_gitaly(finder)
@@ -43,7 +59,7 @@ module Gitlab
       # Headers are added to immitate offset pagination, while it is the default option
       def paginate_first_page_via_gitaly(finder)
         finder.execute(gitaly_pagination: true).tap do |records|
-          total = project.repository.branch_count
+          total = finder.total
           per_page = params[:per_page].presence || Kaminari.config.default_per_page
 
           Gitlab::Pagination::OffsetHeaderBuilder.new(
diff --git a/lib/gitlab/pagination/keyset/column_order_definition.rb b/lib/gitlab/pagination/keyset/column_order_definition.rb
index 0755af9587b..2b968c4253f 100644
--- a/lib/gitlab/pagination/keyset/column_order_definition.rb
+++ b/lib/gitlab/pagination/keyset/column_order_definition.rb
@@ -173,6 +173,18 @@ module Gitlab
           distinct
         end
 
+        def order_direction_as_sql_string
+          sql_string = ascending_order? ? +'ASC' : +'DESC'
+
+          if nulls_first?
+            sql_string << ' NULLS FIRST'
+          elsif nulls_last?
+            sql_string << ' NULLS LAST'
+          end
+
+          sql_string
+        end
+
         private
 
         attr_reader :reversed_order_expression, :nullable, :distinct
diff --git a/lib/gitlab/pagination/keyset/cursor_based_request_context.rb b/lib/gitlab/pagination/keyset/cursor_based_request_context.rb
new file mode 100644
index 00000000000..18390f5b59d
--- /dev/null
+++ b/lib/gitlab/pagination/keyset/cursor_based_request_context.rb
@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Pagination
+    module Keyset
+      class CursorBasedRequestContext
+        DEFAULT_SORT_DIRECTION = :desc
+        attr_reader :request_context
+        delegate :params, to: :request_context
+
+        def initialize(request_context)
+          @request_context = request_context
+        end
+
+        def per_page
+          params[:per_page]
+        end
+
+        def cursor
+          params[:cursor]
+        end
+
+        def apply_headers(cursor_for_next_page)
+          Gitlab::Pagination::Keyset::HeaderBuilder
+            .new(request_context)
+            .add_next_page_header({ cursor: cursor_for_next_page })
+        end
+
+        def order_by
+          { params[:order_by].to_sym => params[:sort]&.to_sym || DEFAULT_SORT_DIRECTION }
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/pagination/keyset/cursor_pager.rb b/lib/gitlab/pagination/keyset/cursor_pager.rb
new file mode 100644
index 00000000000..0b49aa87a02
--- /dev/null
+++ b/lib/gitlab/pagination/keyset/cursor_pager.rb
@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Pagination
+    module Keyset
+      class CursorPager < Gitlab::Pagination::Base
+        attr_reader :cursor_based_request_context, :paginator
+
+        def initialize(cursor_based_request_context)
+          @cursor_based_request_context = cursor_based_request_context
+        end
+
+        def paginate(relation)
+          @paginator ||= relation.keyset_paginate(
+            per_page: cursor_based_request_context.per_page,
+            cursor: cursor_based_request_context.cursor
+          )
+
+          paginator.records
+        end
+
+        def finalize(_records = [])
+          # can be called only after executing `paginate(relation)`
+          apply_headers
+        end
+
+        private
+
+        def apply_headers
+          return unless paginator.has_next_page?
+
+          cursor_based_request_context
+            .apply_headers(paginator.cursor_for_next_page)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/pagination/keyset/in_operator_optimization/array_scope_columns.rb b/lib/gitlab/pagination/keyset/in_operator_optimization/array_scope_columns.rb
new file mode 100644
index 00000000000..95afd5a8595
--- /dev/null
+++ b/lib/gitlab/pagination/keyset/in_operator_optimization/array_scope_columns.rb
@@ -0,0 +1,59 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Pagination
+    module Keyset
+      module InOperatorOptimization
+        class ArrayScopeColumns
+          ARRAY_SCOPE_CTE_NAME = 'array_cte'
+
+          def initialize(columns)
+            validate_columns!(columns)
+
+            array_scope_table = Arel::Table.new(ARRAY_SCOPE_CTE_NAME)
+            @columns = columns.map do |column|
+              ColumnData.new(column, "array_scope_#{column}", array_scope_table)
+            end
+          end
+
+          def array_scope_cte_name
+            ARRAY_SCOPE_CTE_NAME
+          end
+
+          def array_aggregated_columns
+            columns.map(&:array_aggregated_column)
+          end
+
+          def array_aggregated_column_names
+            columns.map(&:array_aggregated_column_name)
+          end
+
+          def arel_columns
+            columns.map(&:arel_column)
+          end
+
+          def array_lookup_expressions_by_position(table_name)
+            columns.map do |column|
+              Arel.sql("#{table_name}.#{column.array_aggregated_column_name}[position]")
+            end
+          end
+
+          private
+
+          attr_reader :columns
+
+          def validate_columns!(columns)
+            if columns.blank?
+              msg = <<~MSG
+              No array columns were given.
+              Make sure you explicitly select the columns in the array_scope parameter.
+              Example: Project.select(:id)
+              MSG
+              raise StandardError, msg
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/pagination/keyset/in_operator_optimization/column_data.rb b/lib/gitlab/pagination/keyset/in_operator_optimization/column_data.rb
new file mode 100644
index 00000000000..3f620f74eca
--- /dev/null
+++ b/lib/gitlab/pagination/keyset/in_operator_optimization/column_data.rb
@@ -0,0 +1,39 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Pagination
+    module Keyset
+      module InOperatorOptimization
+        class ColumnData
+          attr_reader :original_column_name, :as, :arel_table
+
+          def initialize(original_column_name, as, arel_table)
+            @original_column_name = original_column_name.to_s
+            @as = as.to_s
+            @arel_table = arel_table
+          end
+
+          def projection
+            arel_column.as(as)
+          end
+
+          def arel_column
+            arel_table[original_column_name]
+          end
+
+          def arel_column_as
+            arel_table[as]
+          end
+
+          def array_aggregated_column_name
+            "#{arel_table.name}_#{original_column_name}_array"
+          end
+
+          def array_aggregated_column
+            Arel::Nodes::NamedFunction.new('ARRAY_AGG', [arel_column]).as(array_aggregated_column_name)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/pagination/keyset/in_operator_optimization/order_by_columns.rb b/lib/gitlab/pagination/keyset/in_operator_optimization/order_by_columns.rb
new file mode 100644
index 00000000000..d8c69a74e6b
--- /dev/null
+++ b/lib/gitlab/pagination/keyset/in_operator_optimization/order_by_columns.rb
@@ -0,0 +1,76 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Pagination
+    module Keyset
+      module InOperatorOptimization
+        class OrderByColumns
+          include Enumerable
+
+          # This class exposes collection methods for the order by columns
+          #
+          # Example: by modelling the `issues.created_at ASC, issues.id ASC` ORDER BY
+          # SQL clause, this class will receive two ColumnOrderDefinition objects
+          def initialize(columns, arel_table)
+            @columns = columns.map do |column|
+              ColumnData.new(column.attribute_name, "order_by_columns_#{column.attribute_name}", arel_table)
+            end
+          end
+
+          def arel_columns
+            columns.map(&:arel_column)
+          end
+
+          def array_aggregated_columns
+            columns.map(&:array_aggregated_column)
+          end
+
+          def array_aggregated_column_names
+            columns.map(&:array_aggregated_column_name)
+          end
+
+          def original_column_names
+            columns.map(&:original_column_name)
+          end
+
+          def original_column_names_as_arel_string
+            columns.map { |c| Arel.sql(c.original_column_name) }
+          end
+
+          def original_column_names_as_tmp_tamble
+            temp_table = Arel::Table.new('record')
+            original_column_names.map { |c| temp_table[c] }
+          end
+
+          def cursor_values(table_name)
+            columns.each_with_object({}) do |column, hash|
+              hash[column.original_column_name] = Arel.sql("#{table_name}.#{column.array_aggregated_column_name}[position]")
+            end
+          end
+
+          def array_lookup_expressions_by_position(table_name)
+            columns.map do |column|
+              Arel.sql("#{table_name}.#{column.array_aggregated_column_name}[position]")
+            end
+          end
+
+          def replace_value_in_array_by_position_expressions
+            columns.map do |column|
+              name = "#{QueryBuilder::RECURSIVE_CTE_NAME}.#{column.array_aggregated_column_name}"
+              new_value = "next_cursor_values.#{column.original_column_name}"
+              "#{name}[:position_query.position-1]||#{new_value}||#{name}[position_query.position+1:]"
+            end
+          end
+
+          def each(&block)
+            columns.each(&block)
+          end
+
+          private
+
+          attr_reader :columns
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/pagination/keyset/in_operator_optimization/query_builder.rb b/lib/gitlab/pagination/keyset/in_operator_optimization/query_builder.rb
new file mode 100644
index 00000000000..39d6e016ac7
--- /dev/null
+++ b/lib/gitlab/pagination/keyset/in_operator_optimization/query_builder.rb
@@ -0,0 +1,290 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Pagination
+    module Keyset
+      module InOperatorOptimization
+        # rubocop: disable CodeReuse/ActiveRecord
+        class QueryBuilder
+          UnsupportedScopeOrder = Class.new(StandardError)
+
+          RECURSIVE_CTE_NAME = 'recursive_keyset_cte'
+          RECORDS_COLUMN = 'records'
+
+          # This class optimizes slow database queries (PostgreSQL specific) where the
+          # IN SQL operator is used with sorting.
+          #
+          # Arguments:
+          # scope - ActiveRecord::Relation supporting keyset pagination
+          # array_scope - ActiveRecord::Relation for the `IN` subselect
+          # array_mapping_scope - Lambda for connecting scope with array_scope
+          # finder_query - ActiveRecord::Relation for finding one row by the passed in cursor values
+          # values - keyset cursor values (optional)
+          #
+          # Example ActiveRecord query: Issues in the namespace hierarchy
+          # > scope = Issue
+          # >  .where(project_id: Group.find(9970).all_projects.select(:id))
+          # >  .order(:created_at, :id)
+          # >  .limit(20);
+          #
+          # Optimized version:
+          #
+          # > scope = Issue.where({}).order(:created_at, :id) # base scope
+          # > array_scope = Group.find(9970).all_projects.select(:id)
+          # > array_mapping_scope = -> (id_expression) { Issue.where(Issue.arel_table[:project_id].eq(id_expression)) }
+          #
+          # # finding the record by id is good enough, we can ignore the created_at_expression
+          # > finder_query = -> (created_at_expression, id_expression) { Issue.where(Issue.arel_table[:id].eq(id_expression)) }
+          #
+          # > Gitlab::Pagination::Keyset::InOperatorOptimization::QueryBuilder.new(
+          # >   scope: scope,
+          # >   array_scope: array_scope,
+          # >   array_mapping_scope: array_mapping_scope,
+          # >   finder_query: finder_query
+          # > ).execute.limit(20)
+          def initialize(scope:, array_scope:, array_mapping_scope:, finder_query:, values: {})
+            @scope, success = Gitlab::Pagination::Keyset::SimpleOrderBuilder.build(scope)
+
+            unless success
+              error_message = <<~MSG
+              The order on the scope does not support keyset pagination. You might need to define a custom Order object.\n
+              See https://docs.gitlab.com/ee/development/database/keyset_pagination.html#complex-order-configuration\n
+              Or the Gitlab::Pagination::Keyset::Order class for examples
+              MSG
+              raise(UnsupportedScopeOrder, error_message)
+            end
+
+            @order = Gitlab::Pagination::Keyset::Order.extract_keyset_order_object(scope)
+            @array_scope = array_scope
+            @array_mapping_scope = array_mapping_scope
+            @finder_query = finder_query
+            @values = values
+            @model = @scope.model
+            @table_name = @model.table_name
+            @arel_table = @model.arel_table
+          end
+
+          def execute
+            selector_cte = Gitlab::SQL::CTE.new(:array_cte, array_scope)
+
+            cte = Gitlab::SQL::RecursiveCTE.new(RECURSIVE_CTE_NAME, union_args: { remove_duplicates: false, remove_order: false })
+            cte << initializer_query
+            cte << data_collector_query
+
+            q = cte
+              .apply_to(model.where({})
+              .with(selector_cte.to_arel))
+              .select(result_collector_final_projections)
+              .where("count <> 0") # filter out the initializer row
+
+            model.from(q.arel.as(table_name))
+          end
+
+          private
+
+          attr_reader :array_scope, :scope, :order, :array_mapping_scope, :finder_query, :values, :model, :table_name, :arel_table
+
+          def initializer_query
+            array_column_names = array_scope_columns.array_aggregated_column_names + order_by_columns.array_aggregated_column_names
+
+            projections = [
+              *result_collector_initializer_columns,
+              *array_column_names,
+              '0::bigint AS count'
+            ]
+
+            model.select(projections).from(build_column_arrays_query).limit(1)
+          end
+
+          # This query finds the first cursor values for each item in the array CTE.
+          #
+          # array_cte:
+          #
+          # |project_id|
+          # |----------|
+          # |         1|
+          # |         2|
+          # |         3|
+          # |         4|
+          #
+          # For each project_id, find the first issues row by respecting the created_at, id order.
+          #
+          # The `array_mapping_scope` parameter defines how the `array_scope` and the `scope` can be combined.
+          #
+          # scope = Issue.where({}) # empty scope
+          # array_mapping_scope = Issue.where(project_id: X)
+          #
+          # scope.merge(array_mapping_scope) # Issue.where(project_id: X)
+          #
+          # X will be replaced with a value from the `array_cte` temporary table.
+          #
+          # |created_at|id|
+          # |----------|--|
+          # |2020-01-15| 2|
+          # |2020-01-07| 3|
+          # |2020-01-07| 4|
+          # |2020-01-10| 5|
+          def build_column_arrays_query
+            q = Arel::SelectManager.new
+              .project(array_scope_columns.array_aggregated_columns + order_by_columns.array_aggregated_columns)
+              .from(array_cte)
+              .join(Arel.sql("LEFT JOIN LATERAL (#{initial_keyset_query.to_sql}) #{table_name} ON TRUE"))
+
+            order_by_columns.each { |column| q.where(column.arel_column.not_eq(nil)) }
+
+            q.as('array_scope_lateral_query')
+          end
+
+          def array_cte
+            Arel::SelectManager.new
+              .project(array_scope_columns.arel_columns)
+              .from(Arel.sql(array_scope_columns.array_scope_cte_name))
+              .as(array_scope_columns.array_scope_cte_name)
+          end
+
+          def initial_keyset_query
+            keyset_scope = scope.merge(array_mapping_scope.call(*array_scope_columns.arel_columns))
+            order
+              .apply_cursor_conditions(keyset_scope, values, use_union_optimization: true)
+              .reselect(*order_by_columns.arel_columns)
+              .limit(1)
+          end
+
+          def data_collector_query
+            array_column_list = array_scope_columns.array_aggregated_column_names
+
+            order_column_value_arrays = order_by_columns.replace_value_in_array_by_position_expressions
+
+            select = [
+              *result_collector_columns,
+              *array_column_list,
+              *order_column_value_arrays,
+              "#{RECURSIVE_CTE_NAME}.count + 1"
+            ]
+
+            from = <<~SQL
+              #{RECURSIVE_CTE_NAME},
+              #{array_order_query.lateral.as('position_query').to_sql},
+              #{ensure_one_row(next_cursor_values_query).lateral.as('next_cursor_values').to_sql}
+            SQL
+
+            model.select(select).from(from)
+          end
+
+          # NULL guard. This method ensures that NULL values are returned when the passed in scope returns 0 rows.
+          # Example query: returns issues.id or NULL
+          #
+          # SELECT issues.id FROM (VALUES (NULL)) nulls (id)
+          # LEFT JOIN (SELECT id FROM issues WHERE id = 1 LIMIT 1) issues ON TRUE
+          # LIMIT 1
+          def ensure_one_row(query)
+            q = Arel::SelectManager.new
+            q.projections = order_by_columns.original_column_names_as_tmp_tamble
+
+            null_values = [nil] * order_by_columns.count
+
+            from = Arel::Nodes::Grouping.new(Arel::Nodes::ValuesList.new([null_values])).as('nulls')
+
+            q.from(from)
+            q.join(Arel.sql("LEFT JOIN (#{query.to_sql}) record ON TRUE"))
+            q.limit = 1
+            q
+          end
+
+          # This subquery finds the cursor values for the next record by sorting the generated cursor arrays in memory and taking the first element.
+          # It combines the cursor arrays (UNNEST) together and sorts them according to the originally defined ORDER BY clause.
+          #
+          # Example: issues in the group hierarchy with ORDER BY created_at, id
+          #
+          # |project_id|  |created_at|id| # 2 arrays combined: issues_created_at_array, issues_id_array
+          # |----------|  |----------|--|
+          # |         1|  |2020-01-15| 2|
+          # |         2|  |2020-01-07| 3|
+          # |         3|  |2020-01-07| 4|
+          # |         4|  |2020-01-10| 5|
+          #
+          # The query will return the cursor values: (2020-01-07, 3) and the array position: 1
+          # From the position, we can tell that the record belongs to the project with id 2.
+          def array_order_query
+            q = Arel::SelectManager.new
+              .project([*order_by_columns.original_column_names_as_arel_string, Arel.sql('position')])
+              .from("UNNEST(#{list(order_by_columns.array_aggregated_column_names)}) WITH ORDINALITY AS u(#{list(order_by_columns.original_column_names)}, position)")
+
+            order_by_columns.each { |column| q.where(Arel.sql(column.original_column_name).not_eq(nil)) } # ignore rows where all columns are NULL
+
+            q.order(Arel.sql(order_by_without_table_references)).take(1)
+          end
+
+          # This subquery finds the next cursor values after the previously determined position (from array_order_query).
+          # The current cursor values are passed in as SQL literals since the actual values are encoded into SQL arrays.
+          #
+          # Example: issues in the group hierarchy with ORDER BY created_at, id
+          #
+          # |project_id|  |created_at|id| # 2 arrays combined: issues_created_at_array, issues_id_array
+          # |----------|  |----------|--|
+          # |         1|  |2020-01-15| 2|
+          # |         2|  |2020-01-07| 3|
+          # |         3|  |2020-01-07| 4|
+          # |         4|  |2020-01-10| 5|
+          #
+          # Assuming that the determined position is 1, the cursor values will be the following:
+          # - Filter: project_id = 2
+          # - created_at = 2020-01-07
+          # - id = 3
+          def next_cursor_values_query
+            cursor_values = order_by_columns.cursor_values(RECURSIVE_CTE_NAME)
+            array_mapping_scope_columns = array_scope_columns.array_lookup_expressions_by_position(RECURSIVE_CTE_NAME)
+
+            keyset_scope = scope
+              .reselect(*order_by_columns.arel_columns)
+              .merge(array_mapping_scope.call(*array_mapping_scope_columns))
+
+            order
+              .apply_cursor_conditions(keyset_scope, cursor_values, use_union_optimization: true)
+              .reselect(*order_by_columns.arel_columns)
+              .limit(1)
+          end
+
+          # Generates an ORDER BY clause by using the column position index and the original order clauses.
+          # This method is used to sort the collected arrays in SQL.
+          # Example: "issues".created_at DESC , "issues".id ASC => 1 DESC, 2 ASC
+          def order_by_without_table_references
+            order.column_definitions.each_with_index.map do |column_definition, i|
+              "#{i + 1} #{column_definition.order_direction_as_sql_string}"
+            end.join(", ")
+          end
+
+          def result_collector_initializer_columns
+            ["NULL::#{table_name} AS #{RECORDS_COLUMN}"]
+          end
+
+          def result_collector_columns
+            query = finder_query
+              .call(*order_by_columns.array_lookup_expressions_by_position(RECURSIVE_CTE_NAME))
+              .select("#{table_name}")
+              .limit(1)
+
+            ["(#{query.to_sql})"]
+          end
+
+          def result_collector_final_projections
+            ["(#{RECORDS_COLUMN}).*"]
+          end
+
+          def array_scope_columns
+            @array_scope_columns ||= ArrayScopeColumns.new(array_scope.select_values)
+          end
+
+          def order_by_columns
+            @order_by_columns ||= OrderByColumns.new(order.column_definitions, arel_table)
+          end
+
+          def list(array)
+            array.join(', ')
+          end
+        end
+        # rubocop: enable CodeReuse/ActiveRecord
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/pagination/keyset/iterator.rb b/lib/gitlab/pagination/keyset/iterator.rb
index c6f0014a0f4..14807fa37c4 100644
--- a/lib/gitlab/pagination/keyset/iterator.rb
+++ b/lib/gitlab/pagination/keyset/iterator.rb
@@ -6,12 +6,13 @@ module Gitlab
       class Iterator
         UnsupportedScopeOrder = Class.new(StandardError)
 
-        def initialize(scope:, use_union_optimization: true)
+        def initialize(scope:, use_union_optimization: true, in_operator_optimization_options: nil)
           @scope, success = Gitlab::Pagination::Keyset::SimpleOrderBuilder.build(scope)
           raise(UnsupportedScopeOrder, 'The order on the scope does not support keyset pagination') unless success
 
           @order = Gitlab::Pagination::Keyset::Order.extract_keyset_order_object(scope)
-          @use_union_optimization = use_union_optimization
+          @use_union_optimization = in_operator_optimization_options ? false : use_union_optimization
+          @in_operator_optimization_options = in_operator_optimization_options
         end
 
         # rubocop: disable CodeReuse/ActiveRecord
@@ -19,11 +20,10 @@ module Gitlab
           cursor_attributes = {}
 
           loop do
-            current_scope = scope.dup.limit(of)
-            relation = order
-              .apply_cursor_conditions(current_scope, cursor_attributes, { use_union_optimization: @use_union_optimization })
-              .reorder(order)
-              .limit(of)
+            current_scope = scope.dup
+            relation = order.apply_cursor_conditions(current_scope, cursor_attributes, keyset_options)
+            relation = relation.reorder(order) unless @in_operator_optimization_options
+            relation = relation.limit(of)
 
             yield relation
 
@@ -38,6 +38,13 @@ module Gitlab
         private
 
         attr_reader :scope, :order
+
+        def keyset_options
+          {
+            use_union_optimization: @use_union_optimization,
+            in_operator_optimization_options: @in_operator_optimization_options
+          }
+        end
       end
     end
   end
diff --git a/lib/gitlab/pagination/keyset/order.rb b/lib/gitlab/pagination/keyset/order.rb
index ccfa9334a12..80726fc8efd 100644
--- a/lib/gitlab/pagination/keyset/order.rb
+++ b/lib/gitlab/pagination/keyset/order.rb
@@ -152,15 +152,24 @@ module Gitlab
         end
 
         # rubocop: disable CodeReuse/ActiveRecord
-        def apply_cursor_conditions(scope, values = {}, options = { use_union_optimization: false })
+        def apply_cursor_conditions(scope, values = {}, options = { use_union_optimization: false, in_operator_optimization_options: nil })
           values ||= {}
           transformed_values = values.with_indifferent_access
-          scope = apply_custom_projections(scope)
+          scope = apply_custom_projections(scope.dup)
 
           where_values = build_where_values(transformed_values)
 
           if options[:use_union_optimization] && where_values.size > 1
             build_union_query(scope, where_values).reorder(self)
+          elsif options[:in_operator_optimization_options]
+            opts = options[:in_operator_optimization_options]
+
+            Gitlab::Pagination::Keyset::InOperatorOptimization::QueryBuilder.new(
+              **{
+                scope: scope.reorder(self),
+                values: values
+              }.merge(opts)
+            ).execute
           else
             scope.where(build_or_query(where_values)) # rubocop: disable CodeReuse/ActiveRecord
           end
@@ -187,7 +196,7 @@ module Gitlab
           columns = Arel::Nodes::Grouping.new(column_definitions.map(&:column_expression))
           values = Arel::Nodes::Grouping.new(column_definitions.map do |column_definition|
             value = values[column_definition.attribute_name]
-            Arel::Nodes.build_quoted(value, column_definition.column_expression)
+            build_quoted(value, column_definition.column_expression)
           end)
 
           if column_definitions.first.ascending_order?
@@ -197,6 +206,12 @@ module Gitlab
           end
         end
 
+        def build_quoted(value, column_expression)
+          return value if value.instance_of?(Arel::Nodes::SqlLiteral)
+
+          Arel::Nodes.build_quoted(value, column_expression)
+        end
+
         # Adds extra columns to the SELECT clause
         def apply_custom_projections(scope)
           additional_projections = column_definitions.select(&:add_to_projections).map do |column_definition|
@@ -204,7 +219,7 @@ module Gitlab
             column_definition.column_expression.dup.as(column_definition.attribute_name).to_sql
           end
 
-          scope = scope.select(*scope.arel.projections, *additional_projections) if additional_projections
+          scope = scope.reselect(*scope.arel.projections, *additional_projections) unless additional_projections.blank?
           scope
         end
 
diff --git a/lib/gitlab/pagination/offset_pagination.rb b/lib/gitlab/pagination/offset_pagination.rb
index 4f8a6ffb2cc..7b5013f137b 100644
--- a/lib/gitlab/pagination/offset_pagination.rb
+++ b/lib/gitlab/pagination/offset_pagination.rb
@@ -29,7 +29,7 @@ module Gitlab
         return pagination_data unless Feature.enabled?(:api_kaminari_count_with_limit, type: :ops)
 
         limited_total_count = pagination_data.total_count_with_limit
-        if limited_total_count > Kaminari::ActiveRecordRelationMethods::MAX_COUNT_LIMIT
+        if limited_total_count > max_limit
           # The call to `total_count_with_limit` memoizes `@arel` because of a call to `references_eager_loaded_tables?`
           # We need to call `reset` because `without_count` relies on `@arel` being unmemoized
           pagination_data.reset.without_count
@@ -38,6 +38,14 @@ module Gitlab
         end
       end
 
+      def max_limit
+        if Feature.enabled?(:lower_relation_max_count_limit, type: :ops)
+          Kaminari::ActiveRecordRelationMethods::MAX_COUNT_NEW_LOWER_LIMIT
+        else
+          Kaminari::ActiveRecordRelationMethods::MAX_COUNT_LIMIT
+        end
+      end
+
       def needs_pagination?(relation)
         return true unless relation.respond_to?(:current_page)
         return true if params[:page].present? && relation.current_page != params[:page].to_i