1 files changed, 60 insertions, 18 deletions

diff --git a/app/models/namespaces/traversal/linear_scopes.rb b/app/models/namespaces/traversal/linear_scopes.rb
index 2da0e48c2da..f5c44171c42 100644
--- a/app/models/namespaces/traversal/linear_scopes.rb
+++ b/app/models/namespaces/traversal/linear_scopes.rb
@@ -15,12 +15,18 @@ module Namespaces
           select('namespaces.traversal_ids[array_length(namespaces.traversal_ids, 1)] AS id')
         end
 
+        def roots
+          return super unless use_traversal_ids_roots?
+
+          root_ids = all.select("#{quoted_table_name}.traversal_ids[1]").distinct
+          unscoped.where(id: root_ids)
+        end
+
         def self_and_ancestors(include_self: true, hierarchy_order: nil)
           return super unless use_traversal_ids_for_ancestor_scopes?
 
           records = unscoped
-            .without_sti_condition
-            .where(id: without_sti_condition.select('unnest(traversal_ids)'))
+            .where(id: select('unnest(traversal_ids)'))
             .order_by_depth(hierarchy_order)
             .normal_select
 
@@ -40,24 +46,24 @@ module Namespaces
         def self_and_descendants(include_self: true)
           return super unless use_traversal_ids?
 
-          records = self_and_descendants_with_duplicates(include_self: include_self)
-
-          distinct = records.select('DISTINCT on(namespaces.id) namespaces.*')
-
-          distinct.normal_select
+          if Feature.enabled?(:traversal_ids_btree, default_enabled: :yaml)
+            self_and_descendants_with_comparison_operators(include_self: include_self)
+          else
+            records = self_and_descendants_with_duplicates_with_array_operator(include_self: include_self)
+            distinct = records.select('DISTINCT on(namespaces.id) namespaces.*')
+            distinct.normal_select
+          end
         end
 
         def self_and_descendant_ids(include_self: true)
           return super unless use_traversal_ids?
 
-          self_and_descendants_with_duplicates(include_self: include_self)
-            .select('DISTINCT namespaces.id')
-        end
-
-        # Make sure we drop the STI `type = 'Group'` condition for better performance.
-        # Logically equivalent so long as hierarchies remain homogeneous.
-        def without_sti_condition
-          unscope(where: :type)
+          if Feature.enabled?(:traversal_ids_btree, default_enabled: :yaml)
+            self_and_descendants_with_comparison_operators(include_self: include_self).as_ids
+          else
+            self_and_descendants_with_duplicates_with_array_operator(include_self: include_self)
+              .select('DISTINCT namespaces.id')
+          end
         end
 
         def order_by_depth(hierarchy_order)
@@ -75,7 +81,7 @@ module Namespaces
         # When we have queries that break this SELECT * format we can run in to errors.
         # For example `SELECT DISTINCT on(...)` will fail when we chain a `.count` c
         def normal_select
-          unscoped.without_sti_condition.from(all, :namespaces)
+          unscoped.from(all, :namespaces)
         end
 
         private
@@ -84,16 +90,52 @@ module Namespaces
           Feature.enabled?(:use_traversal_ids, default_enabled: :yaml)
         end
 
+        def use_traversal_ids_roots?
+          Feature.enabled?(:use_traversal_ids_roots, default_enabled: :yaml) &&
+          use_traversal_ids?
+        end
+
         def use_traversal_ids_for_ancestor_scopes?
           Feature.enabled?(:use_traversal_ids_for_ancestor_scopes, default_enabled: :yaml) &&
           use_traversal_ids?
         end
 
-        def self_and_descendants_with_duplicates(include_self: true)
+        def self_and_descendants_with_comparison_operators(include_self: true)
+          base = all.select(
+            :traversal_ids,
+            'LEAD (namespaces.traversal_ids, 1) OVER (ORDER BY namespaces.traversal_ids ASC) next_traversal_ids'
+          )
+          cte = Gitlab::SQL::CTE.new(:base_cte, base)
+
+          namespaces = Arel::Table.new(:namespaces)
+          records = unscoped
+            .with(cte.to_arel)
+            .from([cte.table, namespaces])
+
+          # Bound the search space to ourselves (optional) and descendants.
+          #
+          # WHERE (base_cte.next_traversal_ids IS NULL OR base_cte.next_traversal_ids > namespaces.traversal_ids)
+          #   AND next_traversal_ids_sibling(base_cte.traversal_ids) > namespaces.traversal_ids
+          records = records
+            .where(cte.table[:next_traversal_ids].eq(nil).or(cte.table[:next_traversal_ids].gt(namespaces[:traversal_ids])))
+            .where(next_sibling_func(cte.table[:traversal_ids]).gt(namespaces[:traversal_ids]))
+
+          #   AND base_cte.traversal_ids <= namespaces.traversal_ids
+          if include_self
+            records.where(cte.table[:traversal_ids].lteq(namespaces[:traversal_ids]))
+          else
+            records.where(cte.table[:traversal_ids].lt(namespaces[:traversal_ids]))
+          end
+        end
+
+        def next_sibling_func(*args)
+          Arel::Nodes::NamedFunction.new('next_traversal_ids_sibling', args)
+        end
+
+        def self_and_descendants_with_duplicates_with_array_operator(include_self: true)
           base_ids = select(:id)
 
           records = unscoped
-            .without_sti_condition
             .from("namespaces, (#{base_ids.to_sql}) base")
             .where('namespaces.traversal_ids @> ARRAY[base.id]')