diff --git a/lib/gitlab/crawler.rb b/lib/gitlab/crawler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..6a8a0faaec6e351888c0ff353804347738bfcbcb
--- /dev/null
+++ b/lib/gitlab/crawler.rb
@@ -0,0 +1,92 @@
+# frozen_string_literal: true
+
+module Gitlab
+  # Generic crawler class that handles queue management, observer pattern, and basic crawling logic.
+  # This class can be extended for specific crawling needs like relationship crawling, dependency crawling, etc.
+  class Crawler
+    attr_reader :observers, :queue, :processed
+
+    def initialize(observers: nil)
+      @queue = []
+      @processed = Set.new
+      @observers = observers || default_observers
+    end
+
+    # Main crawling method - template method pattern
+    #
+    # Observer events fired during crawling:
+    # - :crawl_start - Called when crawling begins
+    # - :item_crawling - Called before crawling each item (passes the item)
+    # - :item_crawled - Called after crawling each item (passes {item:, data:})
+    # - :crawl_complete - Called when crawling finishes
+    def crawl
+      notify_observers(:crawl_start)
+
+      initialize_queue
+
+      while @queue.any?
+        current = @queue.shift
+        key = item_key(current)
+
+        next if @processed.include?(key)
+
+        @processed.add(key)
+
+        notify_observers(:item_crawling, current)
+
+        data = crawl_item(current)
+
+        notify_observers(:item_crawled, { item: current, data: data })
+
+        next unless data.any?
+
+        process_crawled_item(current, data)
+      end
+
+      notify_observers(:crawl_complete)
+    end
+
+    protected
+
+    # Template methods to be implemented by subclasses
+
+    # Initialize the queue with seed data
+    def initialize_queue
+      raise NotImplementedError, "Subclasses must implement initialize_queue"
+    end
+
+    # Build a unique key for tracking processed items
+    def item_key(queue_item)
+      raise NotImplementedError, "Subclasses must implement item_key"
+    end
+
+    # Crawl data for a queue item
+    def crawl_item(queue_item)
+      raise NotImplementedError, "Subclasses must implement crawl_item"
+    end
+
+    # Process data and potentially add new items to queue
+    def process_crawled_item(queue_item, data)
+      # Default implementation does nothing - subclasses should override
+    end
+
+    # Default observers - can be overridden by subclasses
+    def default_observers
+      []
+    end
+
+    # Add an item to the queue
+    def enqueue(item)
+      @queue << item
+    end
+
+    private
+
+    def notify_observers(event, *args)
+      @observers.each do |observer|
+        method_name = :"on_#{event}"
+        observer.__send__(method_name, *args) if observer.respond_to?(method_name)
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/database/relationship_mapper.rb b/lib/gitlab/database/relationship_mapper.rb
new file mode 100644
index 0000000000000000000000000000000000000000..50d6460925f45b422166190f9ddeb6c31fb0d877
--- /dev/null
+++ b/lib/gitlab/database/relationship_mapper.rb
@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Database
+    class RelationshipMapper
+      def initialize
+        @relationship_sources = [
+          Relationships::ActiveRecord.new,
+          Relationships::ForeignKey.new
+        ]
+      end
+
+      def generate_map
+        Rails.application.eager_load!
+
+        relationships = []
+
+        @relationship_sources.each do |source|
+          relationships.concat(source.extract_relationships)
+        end
+
+        # Sort for consistent output
+        relationships.sort_by do |rel|
+          [rel[:source_model], rel[:target_model], rel[:relationship_name]]
+        end
+      end
+
+      def save_map(output_path = 'config/relationship_map.json')
+        map = generate_map
+        File.write(output_path, Gitlab::Json.pretty_generate(map))
+        map
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/database/foreign_keys.rb b/lib/gitlab/reflections/database/foreign_keys.rb
new file mode 100644
index 0000000000000000000000000000000000000000..adbcc6aa60db68f7069cc62c9be22112d6555809
--- /dev/null
+++ b/lib/gitlab/reflections/database/foreign_keys.rb
@@ -0,0 +1,41 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Database
+      # Application-aware foreign key reflection for GitLab's database schema
+      class ForeignKeys
+        # GitLab application-specific table filtering constants
+        INTERNAL_TABLE_NAMES = %w[schema_migrations ar_internal_metadata].freeze
+
+        SYSTEM_TABLE_PATTERNS = [
+          /^batched_background_migration/,
+          /^postgres_/,
+          /^detached_partitions$/,
+          /^loose_foreign_keys_deleted_records$/,
+          /^ghost_user_migrations$/
+        ].freeze
+
+        class << self
+          def for_relationships
+            scope = ::Gitlab::Database::PostgresForeignKey.not_inherited
+
+            # Filter out internal tables by name
+            INTERNAL_TABLE_NAMES.each do |table_name|
+              scope = scope.where.not(constrained_table_name: table_name)
+                          .where.not(referenced_table_name: table_name)
+            end
+
+            # Filter out system tables by pattern
+            SYSTEM_TABLE_PATTERNS.each do |pattern|
+              scope = scope.where.not("constrained_table_name ~ ?", pattern.source)
+                          .where.not("referenced_table_name ~ ?", pattern.source)
+            end
+
+            scope
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/models/active_record.rb b/lib/gitlab/reflections/models/active_record.rb
new file mode 100644
index 0000000000000000000000000000000000000000..6c9994888a4297bd6156ebf43a3e715b10724b44
--- /dev/null
+++ b/lib/gitlab/reflections/models/active_record.rb
@@ -0,0 +1,71 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Models
+      # ActiveRecord model discovery and analysis for GitLab application
+      class ActiveRecord
+        include Singleton
+
+        attr_reader :table_to_model_index, :model_to_table_index, :models
+
+        def initialize
+          build_indexes
+        end
+
+        private
+
+        def discover_models
+          ::ActiveRecord::Base.descendants
+        end
+
+        def filter_models(models)
+          processed_models = Set.new
+
+          models.reject do |model|
+            skip = skip_model?(model) || processed_models.include?(model.name)
+            processed_models.add(model.name) unless skip
+            skip
+          end
+        end
+
+        def skip_model?(model)
+          return true if model.abstract_class?
+          return true if model.name.nil? # Anonymous models
+          return true if model.table_name.nil?
+          return true if excluded_model?(model)
+
+          false
+        end
+
+        def excluded_model?(model)
+          # Skip test/fixture models, internal Rails models, etc.
+          excluded_patterns = [
+            /\AActiveRecord::/,
+            /\AActiveStorage::/,
+            /Test$/,
+            /Fixture$/,
+            /\AMiniTest::/,
+            /\ARSpec::/
+          ]
+
+          excluded_patterns.any? { |pattern| model.name.match?(pattern) }
+        end
+
+        def build_indexes
+          @table_to_model_index = {}
+          @model_to_table_index = {}
+
+          @models = filter_models(discover_models)
+
+          @models.each do |model|
+            next unless model.table_name
+
+            @table_to_model_index[model.table_name] = model.name
+            @model_to_table_index[model.name] = model.table_name
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/active_record.rb b/lib/gitlab/reflections/relationships/active_record.rb
new file mode 100644
index 0000000000000000000000000000000000000000..a43e136e1ac3a90993832af7227a8fdc28528211
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/active_record.rb
@@ -0,0 +1,106 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      class ActiveRecord < Base
+        SUPPORTED_RELATIONSHIPS = %w[
+          has_many has_one belongs_to has_and_belongs_to_many
+          has_many_attached has_one_attached
+        ].freeze
+
+        RELATIONSHIP_HANDLERS = {
+          'belongs_to' => Handlers::BelongsToHandler,
+          'has_many' => Handlers::HasAssociationHandler,
+          'has_one' => Handlers::HasAssociationHandler,
+          'has_and_belongs_to_many' => Handlers::HabtmHandler,
+          'has_many_attached' => Handlers::ActiveStorageHandler,
+          'has_one_attached' => Handlers::ActiveStorageHandler
+        }.freeze
+
+        def initialize
+          super
+          @models_reflection = Gitlab::Reflections::Models::ActiveRecord.instance
+        end
+
+        private
+
+        def get_source_data
+          @models_reflection.models
+        end
+
+        def filter_source_data(models)
+          # Filtering is now handled by the models reflection
+          models
+        end
+
+        def map_to_relationships(models)
+          relationships = []
+
+          models.each do |model|
+            model.reflections.each do |name, reflection|
+              next unless SUPPORTED_RELATIONSHIPS.include?(reflection.macro.to_s)
+
+              relationship_hashes = build_relationships_for_reflection(model, name, reflection, models)
+              relationships.concat(relationship_hashes) if relationship_hashes
+            end
+          end
+
+          relationships
+        end
+
+        def build_relationships_for_reflection(model, association_name, reflection, all_models)
+          handler_class = determine_handler_class(reflection)
+          return unless handler_class
+
+          handler = handler_class.new(model, association_name, reflection)
+
+          # For polymorphic relationships, we need to discover all possible target tables
+          if handler.polymorphic?
+            build_polymorphic_relationships(handler, all_models)
+          else
+            [handler.build_relationship].compact
+          end
+        end
+
+        def build_polymorphic_relationships(handler, all_models)
+          relationships = []
+
+          # First, add the base polymorphic relationship (with target_table: nil)
+          base_relationship = handler.build_relationship
+          relationships << base_relationship if base_relationship
+
+          # Then, discover all possible target tables for this polymorphic association
+          target_tables = handler.discover_targets(all_models)
+
+          target_tables.each do |target_table|
+            # Create a specific relationship for each target table
+            specific_relationship = handler.build_relationship_with_target(target_table)
+            relationships << specific_relationship if specific_relationship
+          end
+
+          relationships
+        end
+
+        def determine_handler_class(reflection)
+          # Check if this is a polymorphic relationship
+          if polymorphic_belongs_to?(reflection)
+            Handlers::PolymorphicBelongsToHandler
+          elsif polymorphic_has_association?(reflection)
+            Handlers::PolymorphicHasAssociationHandler
+          else
+            RELATIONSHIP_HANDLERS[reflection.macro.to_s]
+          end
+        end
+
+        def polymorphic_belongs_to?(reflection)
+          reflection.macro == :belongs_to && reflection.polymorphic?
+        end
+
+        def polymorphic_has_association?(reflection)
+          [:has_many, :has_one].include?(reflection.macro) && reflection.options[:as]
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/base.rb b/lib/gitlab/reflections/relationships/base.rb
new file mode 100644
index 0000000000000000000000000000000000000000..8e08c97154119f86eb39b0f028eceefc94b97e7d
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/base.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      class Base
+        # Main interface - returns relationship data
+        def extract_relationships
+          source_data = get_source_data
+          filtered_data = filter_source_data(source_data)
+          map_to_relationships(filtered_data)
+        end
+
+        protected
+
+        def build_relationship(**attributes)
+          # Validate required fields
+          required_fields = [:source_table, :target_table, :foreign_key]
+          missing_fields = required_fields.select { |field| attributes[field].nil? || attributes[field].to_s.empty? }
+
+          if missing_fields.any?
+            Rails.logger.debug "Failed to build relationship: missing required fields #{missing_fields}"
+            return
+          end
+
+          # Return clean relationship hash
+          attributes.compact
+        rescue StandardError => e
+          Rails.logger.debug "Failed to build relationship: #{e.message}"
+          nil
+        end
+
+        private
+
+        # Override in subclasses
+        def get_source_data
+          raise NotImplementedError, 'Subclasses must implement get_source_data'
+        end
+
+        def filter_source_data(source_data)
+          raise NotImplementedError, 'Subclasses must implement filter_source_data'
+        end
+
+        def map_to_relationships(filtered_data)
+          raise NotImplementedError, 'Subclasses must implement map_to_relationships'
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/builder.rb b/lib/gitlab/reflections/relationships/builder.rb
new file mode 100644
index 0000000000000000000000000000000000000000..11cdc1c53cf1e7beb3a9804f1e49cf25a879f40f
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/builder.rb
@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      # Builds a relationship array from relationship extractors
+      class Builder
+        def initialize(ar_extractor = nil, fk_extractor = nil)
+          @ar_extractor = ar_extractor || Gitlab::Reflections::Relationships::ActiveRecord.new
+          @fk_extractor = fk_extractor || Gitlab::Reflections::Relationships::ForeignKey.new
+        end
+
+        def build_relationships
+          ar_relationships = @ar_extractor.extract_relationships
+          fk_relationships = @fk_extractor.extract_relationships
+
+          # Combine and transform relationships using pipeline
+          combined_relationships = ar_relationships + fk_relationships
+
+          Transformers::Pipeline.new(
+            Transformers::NormalizeColumns,
+            Transformers::MergeDataSources,
+            Transformers::Deduplicate,
+            Transformers::Validate
+          ).execute(combined_relationships)
+        end
+
+        # Alias for backward compatibility
+        alias_method :build_graph, :build_relationships
+
+        private
+
+        def build_relationship(**attributes)
+          relationship = Relationship.new(**attributes)
+          relationship.valid? ? relationship.to_h : nil
+        rescue ArgumentError => e
+          Rails.logger.debug "Failed to build relationship: #{e.message}"
+          nil
+        end
+
+        # Backward compatibility methods
+        alias_method :build_edge, :build_relationship
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/crawl_logger.rb b/lib/gitlab/reflections/relationships/crawl_logger.rb
new file mode 100644
index 0000000000000000000000000000000000000000..d2cde21fdaa50a08be335ba0619dd479a68e31a5
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/crawl_logger.rb
@@ -0,0 +1,114 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      class CrawlLogger
+        def initialize(seed_table:, seed_id:, timestamp:)
+          @seed_table = seed_table
+          @seed_id = seed_id
+          @timestamp = timestamp
+          @tables_log = Rails.root.join('tmp', "crawl_tables_#{timestamp}.log")
+          @tables_with_data = Set.new
+        end
+
+        def on_crawl_start
+          puts "Starting crawl from #{@seed_table}:#{@seed_id}"
+          puts "Logging tables with data to: #{@tables_log}"
+          puts "=" * 50
+          puts "Validating relationships..."
+        end
+
+        def on_item_crawling(queue_item)
+          puts "Fetching #{queue_item.table} data..."
+        end
+
+        def on_crawl_event(event_data)
+          case event_data[:type]
+          when :relationship_query
+            if event_data[:result_count] > 0
+              log_table_with_data(event_data[:table])
+              puts "✓ Found #{event_data[:result_count]} records in #{event_data[:table]}"
+            else
+              puts "- No data in #{event_data[:table]}"
+            end
+          end
+        end
+
+        def on_crawl_complete(summary_data = nil)
+          puts "=" * 50
+          puts "Crawl completed!"
+
+          if summary_data
+            puts "Tables with data: #{summary_data[:tables_with_data_count]}" if summary_data[:tables_with_data_count]
+            puts "Total queries executed: #{summary_data[:total_queries]}" if summary_data[:total_queries]
+          end
+
+          puts "Tables logged to: #{@tables_log}"
+        end
+
+        def on_validation_error(error_data)
+          puts "⚠ Found #{error_data[:count]} invalid relationships in sample:"
+          error_data[:invalid_relationships].each do |invalid|
+            puts "  [#{invalid[:index]}] #{invalid[:issues].join(', ')}: #{invalid[:relationship].inspect}"
+          end
+        end
+
+        def on_crawl_error(error_data)
+          case error_data[:type]
+          when :query_error
+            puts "ERROR querying #{error_data[:table]}: #{error_data[:message]}"
+          when :table_not_found
+            puts "ERROR: Table '#{error_data[:table]}' does not exist"
+          when :column_not_found
+            puts "ERROR: Column '#{error_data[:column]}' does not exist in table '#{error_data[:table]}'"
+          when :polymorphic_error
+            puts "ERROR processing polymorphic relationship: #{error_data[:message]}"
+          end
+        end
+
+        def on_crawl_skip(skip_data)
+          case skip_data[:type]
+          when :table_ignored
+            puts "SKIPPED: Table '#{skip_data[:table]}' - #{skip_data[:reason]}"
+          end
+        end
+
+        private
+
+        def log_table_with_data(table_name)
+          return if @tables_with_data.include?(table_name)
+
+          @tables_with_data.add(table_name)
+          File.open(@tables_log, 'a') { |f| f.puts table_name }
+        end
+
+        def print_relationship_def(relationship_def)
+          puts "  Input relationship definition:"
+          puts "    #{Gitlab::Json.pretty_generate(relationship_def).gsub(/\n/, "\n    ")}"
+
+          return unless relationship_def&.dig(:source_info)
+
+          source_info = relationship_def[:source_info]
+          puts "  Derived from:"
+
+          if source_info[:has_fk_constraint]
+            source_columns = Array(relationship_def[:source_columns]).join(', ')
+            target_columns = Array(relationship_def[:target_columns]).join(', ')
+            puts "    - Foreign key constraint: #{relationship_def[:source_table]}.[#{source_columns}] -> #{relationship_def[:target_table]}.[#{target_columns}]"
+          end
+
+          if source_info[:forward_association]
+            assoc = source_info[:forward_association]
+            puts "    - ActiveRecord association: #{assoc[:name]} (#{assoc[:type]}) in #{assoc[:class_name] || relationship_def[:source_table].classify}"
+          end
+
+          return unless source_info[:reverse_association]
+
+          assoc = source_info[:reverse_association]
+          puts "    - Reverse association: #{assoc[:name]} (#{assoc[:type]}) in #{assoc[:class_name] || relationship_def[:target_table]&.classify}"
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/crawl_tracer.rb b/lib/gitlab/reflections/relationships/crawl_tracer.rb
new file mode 100644
index 0000000000000000000000000000000000000000..511ba455e27d4f851ce0bde513a26e9f409634f0
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/crawl_tracer.rb
@@ -0,0 +1,90 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      class CrawlTracer
+        def initialize(seed_table:, seed_id:, timestamp:)
+          @seed_table = seed_table
+          @seed_id = seed_id
+          @timestamp = timestamp
+          @debug_log = Rails.root.join('tmp', "crawl_debug_#{timestamp}.log")
+          @sequence_counter = 0
+
+          # Initialize debug log file
+          initialize_debug_log
+        end
+
+        def on_crawl_event(event_data)
+          case event_data[:type]
+          when :relationship_query
+            @sequence_counter += 1
+
+            # Build chain display from relationship_chain if available
+            chain_display = build_chain_display(event_data[:relationship_chain])
+
+            log_debug("RELATIONSHIP_QUERY",
+              "[#{@sequence_counter}] #{event_data[:table]} " \
+                "(#{event_data[:result_count]} records) | " \
+                "Chain: #{chain_display}")
+          end
+        end
+
+        def on_crawl_error(error_data)
+          log_debug("CRAWL_ERROR", "#{error_data[:type]}: #{error_data[:table]} - #{error_data[:message]}")
+        end
+
+        def on_crawl_skip(skip_data)
+          log_debug("CRAWL_SKIP", "#{skip_data[:type]}: #{skip_data[:table]} - #{skip_data[:reason]}")
+        end
+
+        def query_count
+          @sequence_counter
+        end
+
+        private
+
+        def build_chain_display(relationship_chain)
+          return "seed" if relationship_chain.nil? || relationship_chain.empty?
+
+          # Convert ChainStep objects to readable format
+          chain_parts = relationship_chain.map do |step|
+            if step.respond_to?(:to_s)
+              step.to_s  # Uses ChainStep's to_s method: "table.column = value"
+            else
+              step.to_s  # Fallback for any other format
+            end
+          end
+
+          chain_parts.join(" -> ")
+        end
+
+        def initialize_debug_log
+          File.open(@debug_log, 'w') do |f|
+            f.puts "=== DATABASE CRAWL DEBUG LOG ==="
+            f.puts "Seed: #{@seed_table}:#{@seed_id}"
+            f.puts "Started: #{Time.current.strftime('%Y-%m-%d %H:%M:%S')}"
+            f.puts "Log file: #{@debug_log}"
+            f.puts "Use 'tail -f #{@debug_log}' to follow in real-time"
+            f.puts "=" * 50
+            f.flush
+          end
+        rescue StandardError => e
+          Gitlab::AppLogger.warn("Failed to initialize crawl debug log: #{e.message}")
+        end
+
+        def log_debug(event_type, message)
+          timestamp = Time.current.strftime("%H:%M:%S")
+          log_line = "[#{timestamp}] #{event_type}: #{message}"
+
+          File.open(@debug_log, 'a') do |f|
+            f.puts log_line
+            f.flush
+          end
+        rescue StandardError => e
+          Gitlab::AppLogger.warn("Failed to write to crawl debug log: #{e.message}")
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/crawler.rb b/lib/gitlab/reflections/relationships/crawler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..0a2a296678528a092529c66b58c6d1274ed26fc5
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/crawler.rb
@@ -0,0 +1,235 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      # Relationship-specific crawler that extends the generic Crawler
+      # Handles relationship loading, polymorphic relationships, and relationship chain building
+      class Crawler < Gitlab::Crawler
+        # Represents a queue item for relationship crawling
+        QueueItem = Struct.new(
+          :table, :query_conditions, :relationship_chain,
+          keyword_init: true
+        ) do
+          def to_h
+            super.compact
+          end
+        end
+
+        # Represents a step in the relationship chain
+        ChainStep = Struct.new(
+          :relationship, :source_values, :target_values,
+          keyword_init: true
+        ) do
+          def to_s
+            source_table = relationship[:source_table]
+            target_table = relationship[:target_table]
+
+            # Build source part with column=value pairs
+            source_columns = Array(relationship[:source_columns])
+            source_parts = source_columns.zip(source_values).filter_map do |col, val|
+              "#{col}=#{val}" if col && val
+            end
+
+            # Build target part with column=value pairs
+            target_columns = Array(relationship[:target_columns])
+            target_parts = target_columns.zip(target_values).filter_map do |col, val|
+              "#{col}=#{val}" if col && val
+            end
+
+            source_display = source_parts.any? ? "#{source_table}(#{source_parts.join(',')})" : source_table
+            target_display = target_parts.any? ? "#{target_table}(#{target_parts.join(',')})" : target_table
+
+            "#{source_display} -> #{target_display}"
+          end
+        end
+
+        def initialize(
+          relationships_file:, seed_table:, seed_id:, observers: nil, callback_handlers: nil,
+          ignore_tables: nil
+        )
+          @relationships_file = relationships_file
+          @seed_table = seed_table
+          @seed_id = seed_id
+          @timestamp = Time.current.strftime("%Y%m%d_%H%M%S")
+
+          # Load and transform relationships using pipeline
+          raw_relationships = Gitlab::Json.parse(File.read(@relationships_file))
+          # Symbolize keys to maintain consistency with Relationship#to_h output
+          symbolized_relationships = raw_relationships.map(&:deep_symbolize_keys)
+
+          @normalized_relationships = Transformers::Pipeline.new(
+            Transformers::ExpandPolymorphic,
+            Transformers::InvertRelationships,
+            Transformers::FilterThroughRelationships
+          ).execute(symbolized_relationships)
+
+          @tables_with_data = Set.new
+          # Convert to Set for efficient lookup, handle nil gracefully
+          @ignore_tables = Set.new(Array(ignore_tables))
+
+          # Initialize observers (support both old and new parameter names for backwards compatibility)
+          observers = observers || callback_handlers || default_observers
+
+          super(observers: observers)
+        end
+
+        # Access to observers for analysis (backwards compatibility)
+        attr_reader :observers, :ignore_tables
+        alias_method :callback_handlers, :observers
+
+        protected
+
+        def initialize_queue
+          # Initialize queue with seed query
+          enqueue(QueueItem.new(
+            table: @seed_table,
+            query_conditions: { 'id' => @seed_id },
+            relationship_chain: []
+          ))
+        end
+
+        def item_key(queue_item)
+          # Create a unique key representing the entire query
+          conditions_key = queue_item.query_conditions.sort.map { |col, val| "#{col}:#{val}" }.join('&')
+          "#{queue_item.table}:#{conditions_key}"
+        end
+
+        def crawl_item(queue_item)
+          # Build WHERE clause from key/value conditions
+          where_conditions = queue_item.query_conditions.map do |col, val|
+            "#{col} = #{ApplicationRecord.connection.quote(val)}"
+          end
+
+          sql = "SELECT * FROM #{queue_item.table} WHERE #{where_conditions.join(' AND ')}"
+          ApplicationRecord.connection.select_all(sql)
+        rescue StandardError => e
+          notify_observers(:crawl_error, {
+            type: :query_error,
+            table: queue_item.table,
+            message: e.message
+          })
+          []
+        end
+
+        def process_crawled_item(queue_item, records)
+          @tables_with_data.add(queue_item.table)
+
+          notify_observers(:crawl_event, {
+            type: :relationship_query,
+            table: queue_item.table,
+            result_count: records.count,
+            relationship_chain: queue_item.relationship_chain
+          })
+
+          # Queue related tables based on this data
+          records.each do |record|
+            queue_related_tables_from_record(queue_item.table, record, queue_item.relationship_chain)
+          end
+        end
+
+        def default_observers
+          [
+            CrawlLogger.new(seed_table: @seed_table, seed_id: @seed_id, timestamp: @timestamp),
+            CrawlTracer.new(seed_table: @seed_table, seed_id: @seed_id, timestamp: @timestamp)
+          ]
+        end
+
+        # Override enqueue to check for ignored tables
+        def enqueue(item)
+          # Skip ignored tables
+          if should_ignore_table?(item.table)
+            notify_observers(:crawl_skip, {
+              type: :table_ignored,
+              table: item.table,
+              reason: "in ignore list"
+            })
+            return
+          end
+
+          super
+        end
+
+        private
+
+        def should_ignore_table?(table_name)
+          return false if @ignore_tables.empty?
+          return false unless table_name
+
+          @ignore_tables.include?(table_name)
+        end
+
+        def queue_related_tables_from_record(source_table, record, relationship_chain)
+          record_id = record['id']
+          return unless record_id
+
+          # Process all normalized relationships in a single loop
+          @normalized_relationships.select { |rel| rel[:source_table] == source_table }.each do |rel|
+            next if should_ignore_table?(rel[:target_table])
+
+            # Extract values for this relationship
+            relationship_values = extract_relationship_values(rel, record)
+            next if relationship_values.empty?
+
+            queue_relationship(
+              relationship_chain: relationship_chain,
+              relationship: rel,
+              relationship_values: relationship_values
+            )
+          end
+        end
+
+        def extract_relationship_values(relationship, record)
+          source_columns = Array(relationship[:source_columns])
+          target_columns = Array(relationship[:target_columns])
+
+          # Extract source values from the record
+          source_values = source_columns.filter_map { |col| record[col] if record[col] }
+
+          # Return empty if we don't have all required values
+          return {} if source_values.length != source_columns.length
+
+          # Pipeline has already handled polymorphic expansion, so target values
+          # are directly mapped from source values
+          target_values = source_values.dup
+
+          {
+            source_values: source_values,
+            target_values: target_values,
+            source_columns: source_columns,
+            target_columns: target_columns
+          }
+        end
+
+        def queue_relationship(relationship_chain:, relationship:, relationship_values:)
+          target_table = relationship[:target_table]
+          target_columns = relationship_values[:target_columns]
+          target_values = relationship_values[:target_values]
+
+          return if target_columns.empty? || target_values.empty?
+
+          # Build query conditions as key/value pairs
+          query_conditions = target_columns.zip(target_values).to_h
+
+          # Add any additional conditions from the relationship (e.g., polymorphic type constraints)
+          query_conditions.merge!(relationship[:conditions]) if relationship[:conditions]
+
+          # Build relationship step
+          relationship_step = ChainStep.new(
+            relationship: relationship,
+            source_values: relationship_values[:source_values],
+            target_values: relationship_values[:target_values]
+          )
+
+          new_chain = relationship_chain + [relationship_step]
+
+          enqueue(QueueItem.new(
+            table: target_table,
+            query_conditions: query_conditions,
+            relationship_chain: new_chain
+          ))
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/foreign_key.rb b/lib/gitlab/reflections/relationships/foreign_key.rb
new file mode 100644
index 0000000000000000000000000000000000000000..587d7c044d4bfb4220920c9be91e410e64b4c8a8
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/foreign_key.rb
@@ -0,0 +1,72 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      class ForeignKey < Base
+        def initialize(fk_reflection = nil)
+          super()
+          @fk_reflection = fk_reflection || Gitlab::Reflections::Database::ForeignKeys
+        end
+
+        private
+
+        def get_source_data
+          table_foreign_keys = []
+
+          @fk_reflection.for_relationships.find_each do |pg_fk|
+            # Convert PostgresForeignKey to our internal format
+            table_foreign_keys << {
+              table_name: pg_fk.constrained_table_name,
+              foreign_key: build_fk_definition_from_pg_fk(pg_fk)
+            }
+          end
+
+          table_foreign_keys
+        end
+
+        def filter_source_data(table_foreign_keys)
+          # Filtering is already done by the fk_reflection scope
+          table_foreign_keys
+        end
+
+        def map_to_relationships(filtered_items)
+          filtered_items.filter_map do |item|
+            build_foreign_key_relationship(item[:table_name], item[:foreign_key])
+          end
+        end
+
+        def build_foreign_key_relationship(table_name, foreign_key)
+          build_relationship(
+            source_table: foreign_key.to_table,
+            target_table: table_name,
+            relationship_type: 'one_to_many',
+            foreign_key: foreign_key.column,
+            foreign_key_table: table_name,
+            source_columns: [foreign_key.primary_key || 'id'],
+            target_columns: [foreign_key.column],
+            has_fk_constraint: true,
+            fk_metadata: {
+              constraint_name: foreign_key.name,
+              on_delete: foreign_key.on_delete,
+              on_update: foreign_key.on_update
+            },
+            data_sources: ['foreign_key']
+          )
+        end
+
+        # Convert PostgresForeignKey to ActiveRecord::ConnectionAdapters::ForeignKeyDefinition-like object
+        def build_fk_definition_from_pg_fk(pg_fk)
+          OpenStruct.new(
+            column: pg_fk.constrained_columns.first, # Assuming single column FK for now
+            to_table: pg_fk.referenced_table_name,
+            primary_key: pg_fk.referenced_columns.first,
+            name: pg_fk.name,
+            on_delete: pg_fk.on_delete_action,
+            on_update: pg_fk.on_update_action
+          )
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/handlers/active_storage_handler.rb b/lib/gitlab/reflections/relationships/handlers/active_storage_handler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..e2408e83a71459bc8e914586a35c45a72633eeae
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/handlers/active_storage_handler.rb
@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Handlers
+        class ActiveStorageHandler < BaseHandler
+          def build_relationship
+            build_relationship_hash(
+              **base_attributes,
+              foreign_key: 'record_id',
+              foreign_key_table: 'active_storage_attachments'
+            )
+          end
+
+          private
+
+          def determine_target_table
+            '**************************'
+          end
+
+          def base_attributes
+            # Override to set the correct relationship type for Active Storage
+            super.merge(
+              relationship_type: reflection.macro == :has_many_attached ? 'one_to_many' : 'one_to_one'
+            )
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/handlers/base_handler.rb b/lib/gitlab/reflections/relationships/handlers/base_handler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..6fcadd1c14e1a1a591bea2c240822ea04f24e8c9
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/handlers/base_handler.rb
@@ -0,0 +1,85 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Handlers
+        class BaseHandler
+          def initialize(model, association_name, reflection)
+            @model = model
+            @association_name = association_name
+            @reflection = reflection
+          end
+
+          def build_relationship
+            raise NotImplementedError, 'Subclasses must implement build_relationship'
+          end
+
+          def build_relationship_with_target(target_table)
+            # Default implementation for non-polymorphic relationships
+            build_relationship
+          end
+
+          def polymorphic?
+            false
+          end
+
+          attr_reader :model, :association_name, :reflection
+
+          private
+
+          def build_relationship_hash(**attributes)
+            required_fields = [:source_table, :target_table, :foreign_key]
+            missing_fields = required_fields.select { |field| attributes[field].nil? || attributes[field].to_s.empty? }
+
+            if missing_fields.any?
+              Gitlab::AppLogger.debug "Failed to build relationship: missing required fields #{missing_fields}"
+              return
+            end
+
+            # Return clean relationship hash
+            attributes.compact
+          rescue StandardError => e
+            Gitlab::AppLogger.debug "Failed to build relationship: #{e.message}"
+            nil
+          end
+
+          def determine_target_table
+            reflection.klass.table_name
+          rescue NameError, ArgumentError
+            reflection.table_name || reflection.plural_name
+          end
+
+          def base_attributes
+            {
+              source_table: model.table_name,
+              target_table: determine_target_table,
+              relationship_type: determine_relationship_type,
+              forward_association: {
+                name: association_name.to_s,
+                type: reflection.macro.to_s,
+                model: model.name
+              },
+              has_ar_forward: true,
+              polymorphic: false,
+              data_sources: ['active_record']
+            }
+          end
+
+          def determine_relationship_type
+            case reflection.macro
+            when :has_many, :has_and_belongs_to_many
+              'one_to_many'
+            when :has_one
+              'one_to_one'
+            when :belongs_to
+              'many_to_one'
+            else
+              'unknown'
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/handlers/belongs_to_handler.rb b/lib/gitlab/reflections/relationships/handlers/belongs_to_handler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..f8fd41b321c5cef0b0c23221a8d9655544beb844
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/handlers/belongs_to_handler.rb
@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Handlers
+        class BelongsToHandler < BaseHandler
+          def build_relationship
+            build_relationship_hash(
+              **base_attributes,
+              foreign_key: reflection.foreign_key,
+              foreign_key_table: model.table_name
+            )
+          end
+
+          def build_relationship_with_target(_target_table)
+            # Non-polymorphic belongs_to handlers don't need target-specific relationships
+            build_relationship
+          end
+
+          private
+
+          def base_attributes
+            # For belongs_to, we need to swap source/target to match database FK direction
+            # The referenced table (parent) should be source, referencing table (child) should be target
+            {
+              source_table: determine_target_table,  # The referenced table (e.g., namespaces)
+              target_table: model.table_name,        # The referencing table (e.g., projects)
+              source_columns: [reflection.association_primary_key || 'id'],
+              target_columns: [reflection.foreign_key],
+              relationship_type: 'many_to_one',
+              reverse_association: {
+                name: association_name.to_s,
+                type: reflection.macro.to_s,
+                model: model.name
+              },
+              has_ar_reverse: true,
+              polymorphic: false,
+              data_sources: ['active_record']
+            }
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/handlers/habtm_handler.rb b/lib/gitlab/reflections/relationships/handlers/habtm_handler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..a2aa68fb54a73abe9fd6e26b8cf958798c91fcd7
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/handlers/habtm_handler.rb
@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Handlers
+        class HabtmHandler < BaseHandler
+          def build_relationship
+            build_relationship_hash(
+              **base_attributes,
+              foreign_key: reflection.foreign_key,
+              foreign_key_table: reflection.join_table,
+              through_table: reflection.join_table,
+              through_target_key: reflection.association_foreign_key
+            )
+          end
+
+          private
+
+          def base_attributes
+            # Override to set the correct relationship type for HABTM
+            super.merge(relationship_type: 'many_to_many')
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/handlers/has_association_handler.rb b/lib/gitlab/reflections/relationships/handlers/has_association_handler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..1d10695c60bcbc70e13defdae3739ced94b006a0
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/handlers/has_association_handler.rb
@@ -0,0 +1,67 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Handlers
+        class HasAssociationHandler < BaseHandler
+          def build_relationship
+            attributes = base_attributes.merge(
+              foreign_key: determine_foreign_key,
+              foreign_key_table: determine_foreign_key_table,
+              source_columns: [(reflection.options[:primary_key] || reflection.association_primary_key || 'id').to_s],
+              target_columns: [determine_foreign_key],
+              through_table: determine_through_table,
+              through_target_key: determine_through_target_key,
+              is_through_association: reflection.through_reflection?
+            )
+
+            build_relationship_hash(**attributes)
+          end
+
+          def build_relationship_with_target(_target_table)
+            # Non-polymorphic has_association handlers don't need target-specific relationships
+            build_relationship
+          end
+
+          private
+
+          def determine_foreign_key
+            if reflection.through_reflection
+              reflection.source_reflection&.foreign_key
+            else
+              reflection.foreign_key
+            end
+          end
+
+          def determine_foreign_key_table
+            if reflection.through_reflection
+              reflection.through_reflection.table_name
+            else
+              determine_target_table
+            end
+          rescue StandardError
+            nil
+          end
+
+          def determine_through_table
+            return unless reflection.through_reflection?
+
+            case reflection.through_reflection.macro
+            when :has_and_belongs_to_many
+              reflection.through_reflection.join_table
+            else
+              reflection.through_reflection.table_name
+            end
+          end
+
+          def determine_through_target_key
+            return unless reflection.through_reflection?
+
+            reflection.source_reflection&.foreign_key
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/handlers/polymorphic_belongs_to_handler.rb b/lib/gitlab/reflections/relationships/handlers/polymorphic_belongs_to_handler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..6616f3c8800900f9d548df037476d858e1065bae
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/handlers/polymorphic_belongs_to_handler.rb
@@ -0,0 +1,90 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Handlers
+        class PolymorphicBelongsToHandler < BelongsToHandler
+          def build_relationship
+            build_relationship_hash(
+              **base_polymorphic_attributes,
+              foreign_key: reflection.foreign_key,
+              foreign_key_table: model.table_name
+            )
+          end
+
+          def build_relationship_with_target(target_table)
+            build_relationship_hash(
+              **base_attributes_with_target(target_table),
+              foreign_key: reflection.foreign_key,
+              foreign_key_table: model.table_name
+            )
+          end
+
+          def polymorphic?
+            true
+          end
+
+          def discover_targets(all_models)
+            target_tables = []
+            polymorphic_macros = [:has_many, :has_one].freeze
+            polymorphic_name = association_name.to_s
+
+            all_models.each do |model|
+              model.reflections.each_value do |reflection|
+                if polymorphic_macros.include?(reflection.macro) &&
+                    reflection.options[:as]&.to_s == polymorphic_name
+                  target_tables << model.table_name
+                end
+              end
+            end
+
+            target_tables.uniq
+          end
+
+          private
+
+          def base_polymorphic_attributes
+            {
+              source_table: model.table_name,
+              target_table: nil, # Polymorphic relationships don't have a single target
+              relationship_type: 'many_to_one',
+              reverse_association: {
+                name: association_name.to_s,
+                type: reflection.macro.to_s,
+                model: model.name
+              },
+              has_ar_reverse: true,
+              polymorphic: true,
+              polymorphic_type_column: polymorphic_type_column,
+              polymorphic_name: association_name.to_s,
+              data_sources: ['active_record']
+            }
+          end
+
+          def base_attributes_with_target(target_table)
+            {
+              source_table: target_table,
+              target_table: model.table_name,
+              relationship_type: 'one_to_many',
+              forward_association: {
+                name: association_name.to_s,
+                type: 'has_many',
+                model: target_table.classify
+              },
+              has_ar_forward: true,
+              polymorphic: true,
+              polymorphic_type_column: polymorphic_type_column,
+              polymorphic_name: association_name.to_s,
+              data_sources: ['active_record']
+            }
+          end
+
+          def polymorphic_type_column
+            reflection.foreign_type || "#{association_name}_type"
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/handlers/polymorphic_has_association_handler.rb b/lib/gitlab/reflections/relationships/handlers/polymorphic_has_association_handler.rb
new file mode 100644
index 0000000000000000000000000000000000000000..ceb3a751fccfdc55c07852812d340130028e90f7
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/handlers/polymorphic_has_association_handler.rb
@@ -0,0 +1,106 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Handlers
+        class PolymorphicHasAssociationHandler < HasAssociationHandler
+          def build_relationship
+            attributes = base_polymorphic_attributes.merge(
+              foreign_key: determine_foreign_key,
+              foreign_key_table: determine_foreign_key_table,
+              source_columns: [reflection.association_primary_key || 'id'],
+              target_columns: [determine_foreign_key],
+              through_table: determine_through_table,
+              through_target_key: determine_through_target_key,
+              is_through_association: reflection.through_reflection?
+            )
+
+            build_relationship_hash(**attributes)
+          end
+
+          def build_relationship_with_target(target_table)
+            attributes = base_attributes_with_target(target_table).merge(
+              foreign_key: determine_foreign_key,
+              foreign_key_table: determine_foreign_key_table,
+              source_columns: [reflection.association_primary_key || 'id'],
+              target_columns: [determine_foreign_key],
+              through_table: determine_through_table,
+              through_target_key: determine_through_target_key,
+              is_through_association: reflection.through_reflection?
+            )
+
+            build_relationship_hash(**attributes)
+          end
+
+          def polymorphic?
+            true
+          end
+
+          def discover_targets(all_models)
+            target_tables = []
+            as_name = reflection.options[:as].to_s
+
+            all_models.each do |model|
+              model.reflections.each do |name, reflection|
+                next unless reflection.macro == :belongs_to &&
+                  reflection.polymorphic? &&
+                  name.to_s == as_name
+
+                target_tables << model.table_name
+              end
+            end
+
+            target_tables.uniq
+          end
+
+          private
+
+          def base_polymorphic_attributes
+            {
+              source_table: model.table_name,
+              target_table: nil, # Polymorphic relationships don't have a single target
+              relationship_type: determine_relationship_type,
+              forward_association: {
+                name: association_name.to_s,
+                type: reflection.macro.to_s,
+                model: model.name
+              },
+              has_ar_forward: true,
+              polymorphic: true,
+              polymorphic_type_column: polymorphic_type_column,
+              polymorphic_name: polymorphic_association_name,
+              data_sources: ['active_record']
+            }
+          end
+
+          def base_attributes_with_target(target_table)
+            {
+              source_table: model.table_name,
+              target_table: target_table,
+              relationship_type: determine_relationship_type,
+              forward_association: {
+                name: association_name.to_s,
+                type: reflection.macro.to_s,
+                model: model.name
+              },
+              has_ar_forward: true,
+              polymorphic: true,
+              polymorphic_type_column: polymorphic_type_column,
+              polymorphic_name: polymorphic_association_name,
+              data_sources: ['active_record']
+            }
+          end
+
+          def polymorphic_type_column
+            "#{reflection.options[:as]}_type"
+          end
+
+          def polymorphic_association_name
+            reflection.options[:as].to_s
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/relationship.rb b/lib/gitlab/reflections/relationships/relationship.rb
new file mode 100644
index 0000000000000000000000000000000000000000..973f0d258609b9566160153dbbf9341ba023d43c
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/relationship.rb
@@ -0,0 +1,222 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      # Represents a single relationship in the relationship collection between database tables.
+      # A relationship captures the raw/primitive relationship metadata directly extracted from
+      # ActiveRecord reflections and database foreign key constraints.
+      #
+      # This class stores ONLY primitive parsed data - no derived or processed information.
+      # Processed data like relationship types, polymorphic targets, etc. should be computed
+      # in later processing steps.
+      #
+      # == Attribute Relationships and Direction
+      #
+      # The source/target and foreign_key attributes serve different purposes and are NOT redundant:
+      #
+      # - source_table/source_columns: The table/columns being referenced (usually the "parent" with primary key)
+      # - target_table/target_columns: The table/columns doing the referencing (usually the "child" with foreign key)
+      # - foreign_key/foreign_key_table: The actual physical foreign key column and which table it's in
+      #
+      # Example for users → posts relationship:
+      #   source_table: 'users', source_columns: ['id'] (the referenced primary key)
+      #   target_table: 'posts', target_columns: ['user_id'] (the referencing column)
+      #   foreign_key: 'user_id', foreign_key_table: 'posts' (the actual FK constraint)
+      #
+      # Example for polymorphic notes → issues relationship:
+      #   source_table: 'notes', source_columns: ['noteable_id', 'noteable_type']
+      #   target_table: 'issues', target_columns: ['id', 'Issue']
+      #   foreign_key: 'noteable_id', foreign_key_table: 'notes'
+      #
+      # This distinction matters for:
+      # - Through relationships: FK might be in an intermediate table
+      # - Complex joins: FK column name might differ from target column name
+      # - Polymorphic associations: Multiple target tables can share the same FK
+      #
+      # == Direction and Bidirectional Information
+      #
+      # Direction is determined by data source and association type. The system creates
+      # inverted information by parsing relationships from multiple sources:
+      #
+      # 1. ActiveRecord associations create relationships based on the model defining the association:
+      #    - User.has_many :posts creates forward relationship (users → posts) with has_ar_forward: true
+      #    - Post.belongs_to :user creates reverse relationship (users → posts) with has_ar_reverse: true
+      #    Note: Both associations describe the same relationship direction (users → posts)
+      #    but from different model perspectives
+      #
+      # 2. Foreign Key constraints always create relationships from referenced → referencing table:
+      #    - FK constraint posts.user_id → users.id creates relationship (users → posts)
+      #
+      # 3. Collection merges relationships with same key (source_table, target_table, source_columns, target_columns):
+      #    - When both User.has_many :posts and Post.belongs_to :user exist, they merge into
+      #      one bidirectional relationship with both has_ar_forward: true and has_ar_reverse: true
+      #
+      # @attr source_table [String] The name of the source table in the relationship
+      # @attr target_table [String] The name of the target table in the relationship
+      # @attr source_columns [Array<String>] The columns in the source table that are referenced
+      #   (supports composite keys and polymorphic types)
+      # @attr target_columns [Array<String>] The columns in the target table that reference the source
+      #   (supports composite keys and polymorphic types)
+      # @attr foreign_key [String] The foreign key column name that establishes the relationship
+      # @attr foreign_key_table [String] The table that contains the foreign key column
+      # @attr relationship_type [String] The type of relationship directly derived from ActiveRecord macro (one_to_one, one_to_many, many_to_one, many_to_many)
+      # @attr forward_association [Hash] ActiveRecord association metadata for the forward direction (source -> target)
+      # @attr reverse_association [Hash] ActiveRecord association metadata for the reverse direction (target -> source)
+      # @attr through_table [String] The intermediate table name for many-to-many relationships
+      # @attr through_target_key [String] The key column in the through table that points to the target
+      # @attr polymorphic [Boolean] Whether this is a polymorphic association
+      # @attr polymorphic_type_column [String] The column name that stores the polymorphic type
+      # @attr polymorphic_name [String] The name of the polymorphic association (e.g., 'commentable', 'notable')
+      # @attr has_ar_forward [Boolean] Whether ActiveRecord defines a forward association for this relationship
+      # @attr has_ar_reverse [Boolean] Whether ActiveRecord defines a reverse association for this relationship
+      # @attr has_fk_constraint [Boolean] Whether a foreign key constraint exists in the database
+      # @attr fk_metadata [Hash] Additional metadata about the foreign key constraint
+      # @attr data_sources [Array<String>] List of data sources that contributed to discovering this relationship
+      # @attr conditions [Hash] Additional conditions that must be met for this relationship (e.g., polymorphic type constraints)
+
+      class Relationship
+        include ActiveModel::Model
+
+        VALID_RELATIONSHIP_TYPES = %w[
+          one_to_one one_to_many many_to_one many_to_many
+        ].freeze
+
+        VALID_ASSOCIATION_TYPES = %w[
+          has_many has_one belongs_to has_and_belongs_to_many
+          has_many_attached has_one_attached
+        ].freeze
+
+        attr_accessor :source_table, :target_table, :foreign_key, :foreign_key_table,
+          :source_columns, :target_columns,
+          :relationship_type, :forward_association, :reverse_association,
+          :through_table, :through_target_key, :polymorphic, :polymorphic_type_column,
+          :polymorphic_name, :has_ar_forward, :has_ar_reverse,
+          :has_fk_constraint, :fk_metadata, :data_sources, :is_through_association,
+          :conditions
+
+        validates :source_table, :foreign_key, presence: true
+        validates :target_table, presence: true, unless: :polymorphic?
+        validates :relationship_type, inclusion: { in: VALID_RELATIONSHIP_TYPES }, allow_blank: true
+        validate :valid_associations?
+
+        def initialize(attributes = {})
+          # Ensure data_sources is always an array
+          attributes = attributes.dup
+          attributes[:data_sources] = Array(attributes[:data_sources]) if attributes.key?(:data_sources)
+
+          # Set default values
+          attributes[:polymorphic] = false unless attributes.key?(:polymorphic)
+          attributes[:has_ar_forward] = false unless attributes.key?(:has_ar_forward)
+          attributes[:has_ar_reverse] = false unless attributes.key?(:has_ar_reverse)
+          attributes[:has_fk_constraint] = false unless attributes.key?(:has_fk_constraint)
+          attributes[:is_through_association] = false unless attributes.key?(:is_through_association)
+          attributes[:data_sources] = [] unless attributes.key?(:data_sources)
+
+          # Set default source_columns to ['id'] if not provided
+          attributes[:source_columns] ||= ['id'] unless attributes.key?(:source_columns)
+
+          super
+        end
+
+        def through_relationship?
+          !!(through_table && !through_table.empty?)
+        end
+
+        def polymorphic?
+          polymorphic
+        end
+
+        # Returns all possible target models for this polymorphic relationship
+        # Note: This should be computed in a later processing step by analyzing
+        # all relationships with the same polymorphic_type_column
+        def polymorphic_target_models
+          []
+        end
+
+        # Returns the polymorphic association name (e.g., 'commentable', 'notable')
+        def polymorphic_association_name
+          return unless polymorphic?
+
+          polymorphic_name
+        end
+
+        # Returns the type column name for polymorphic relationships
+        def polymorphic_type_column_name
+          return unless polymorphic?
+
+          polymorphic_type_column
+        end
+
+        # Hash-style attribute access - supports both symbol and string keys
+        def [](attr_name)
+          # Convert string keys to symbols for consistency
+          attr_name = attr_name.to_sym if attr_name.is_a?(String)
+          public_send(attr_name)
+        end
+
+        def to_h
+          {
+            source_table: source_table,
+            target_table: target_table,
+            source_columns: source_columns,
+            target_columns: target_columns,
+            foreign_key: foreign_key,
+            foreign_key_table: foreign_key_table,
+            relationship_type: relationship_type,
+            forward_association: forward_association,
+            reverse_association: reverse_association,
+            through_table: through_table,
+            through_target_key: through_target_key,
+            polymorphic: polymorphic,
+            polymorphic_type_column: polymorphic_type_column,
+            polymorphic_name: polymorphic_name,
+            has_ar_forward: has_ar_forward,
+            has_ar_reverse: has_ar_reverse,
+            has_fk_constraint: has_fk_constraint,
+            fk_metadata: fk_metadata,
+            data_sources: data_sources,
+            is_through_association: is_through_association,
+            conditions: conditions
+          }.compact
+        end
+
+        def ==(other)
+          return false unless other.is_a?(self.class)
+
+          to_h == other.to_h
+        end
+
+        alias_method :eql?, :==
+
+        def hash
+          to_h.hash
+        end
+
+        def inspect
+          "#<#{self.class.name} #{source_table} -> #{target_table} (#{relationship_type})>"
+        end
+
+        private
+
+        def valid_associations?
+          valid_forward = forward_association.nil? || valid_association?(forward_association)
+          valid_reverse = reverse_association.nil? || valid_association?(reverse_association)
+
+          errors.add(:base, 'Invalid association types') unless valid_forward && valid_reverse
+
+          valid_forward && valid_reverse
+        end
+
+        def valid_association?(association)
+          return false unless association.is_a?(Hash)
+
+          return false unless association[:name] && !association[:name].empty? &&
+            association[:type] && !association[:type].empty?
+
+          VALID_ASSOCIATION_TYPES.include?(association[:type])
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/transformers/deduplicate.rb b/lib/gitlab/reflections/relationships/transformers/deduplicate.rb
new file mode 100644
index 0000000000000000000000000000000000000000..8248030ecf392dd8ea9553d31c0ea117de7b3132
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/deduplicate.rb
@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        class Deduplicate
+          def self.call(relationships)
+            new(relationships).transform
+          end
+
+          def initialize(relationships)
+            @relationships = relationships
+          end
+
+          def transform
+            seen = Set.new
+            @relationships.filter_map do |rel|
+              key = relationship_signature(rel)
+              if seen.include?(key)
+                nil
+              else
+                seen.add(key)
+                rel
+              end
+            end
+          end
+
+          private
+
+          def relationship_signature(rel)
+            # Create a signature that uniquely identifies a relationship
+            [
+              rel[:source_table],
+              rel[:target_table],
+              Array(rel[:source_columns]).sort.join(','),
+              Array(rel[:target_columns]).sort.join(','),
+              rel[:relationship_type],
+              rel[:polymorphic_type_value]
+            ].compact.join('|')
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/transformers/expand_polymorphic.rb b/lib/gitlab/reflections/relationships/transformers/expand_polymorphic.rb
new file mode 100644
index 0000000000000000000000000000000000000000..82da1a65bf143f289c7f6a5e5efdb2b528f7c78a
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/expand_polymorphic.rb
@@ -0,0 +1,120 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        # Transforms polymorphic relationships into concrete has_many relationships.
+        #
+        # This transformer converts a single polymorphic relationship like:
+        #   User.has_many :notes, as: :noteable
+        #
+        # Into multiple concrete has_many relationships with conditions:
+        #   User.has_many :issue_notes, -> { where(noteable_type: 'Issue') }, class_name: 'Note'
+        #   User.has_many :merge_request_notes, -> { where(noteable_type: 'MergeRequest') }, class_name: 'Note'
+        #
+        # Each expanded relationship includes a 'conditions' attribute that specifies
+        # the polymorphic type constraint: { 'noteable_type' => 'Issue' }
+        class ExpandPolymorphic
+          def self.call(relationships)
+            new(relationships).transform
+          end
+
+          def initialize(relationships)
+            @relationships = relationships
+          end
+
+          def transform
+            expanded = []
+
+            @relationships.each do |rel|
+              if rel[:polymorphic] == true && rel[:target_table].nil?
+                # Convert polymorphic relationship into multiple has_many relationships
+                expanded.concat(expand_into_has_many_relationships(rel))
+              else
+                # Pass through non-polymorphic relationships and concrete polymorphic relationships
+                expanded << rel
+              end
+            end
+
+            expanded
+          end
+
+          private
+
+          # Converts a single polymorphic relationship into multiple concrete has_many relationships
+          def expand_into_has_many_relationships(rel)
+            target_tables = find_polymorphic_targets(rel)
+
+            target_tables.map do |target_table|
+              type_column = rel[:polymorphic_type_column] || infer_type_column(rel)
+              type_value = target_table.classify
+
+              # Create a concrete has_many relationship with polymorphic condition
+              concrete_relationship = rel.merge(
+                target_table: target_table,
+                relationship_type: 'one_to_many',  # Convert to one_to_many
+                polymorphic: false,                # No longer polymorphic
+                polymorphic_type_value: type_value,
+                polymorphic_type_column: type_column,
+                conditions: { type_column => type_value }
+              )
+
+              add_polymorphic_columns(concrete_relationship, type_column, type_value)
+            end
+          end
+
+          def add_polymorphic_columns(relationship, type_column, type_value)
+            # Ensure source_columns and target_columns are arrays (make copies to avoid mutation)
+            source_columns = Array(relationship[:source_columns]).dup
+            target_columns = Array(relationship[:target_columns]).dup
+
+            # Add the type column to source_columns and type_value to target_columns
+            type_column_index = source_columns.index(type_column)
+
+            if type_column_index
+              # Replace the corresponding target_columns value with the type_value
+              target_columns[type_column_index] = type_value
+            else
+              # Add type column/value to the column arrays if not already present
+              source_columns += [type_column]
+              target_columns += [type_value]
+            end
+
+            relationship.merge(
+              source_columns: source_columns,
+              target_columns: target_columns
+            )
+          end
+
+          def find_polymorphic_targets(rel)
+            # Look through all relationships to find ones that target this polymorphic
+            polymorphic_name = rel[:polymorphic_name]
+            targets = Set.new
+
+            @relationships.each do |other_rel|
+              if other_rel[:polymorphic_name] == polymorphic_name && other_rel[:target_table]
+                targets << other_rel[:target_table]
+              end
+            end
+
+            targets.to_a
+          end
+
+          def infer_type_column(rel)
+            # Infer the type column name from the polymorphic association
+            # Standard Rails convention: if foreign key is 'commentable_id', type column is 'commentable_type'
+            foreign_key = rel[:foreign_key]
+            return unless foreign_key
+
+            if foreign_key.end_with?('_id')
+              foreign_key.sub(/_id$/, '_type')
+            else
+              "#{foreign_key}_type"
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/transformers/filter_tables.rb b/lib/gitlab/reflections/relationships/transformers/filter_tables.rb
new file mode 100644
index 0000000000000000000000000000000000000000..6f8fedc51250dec5d9309e5c10b582a5272a506e
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/filter_tables.rb
@@ -0,0 +1,48 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        class FilterTables
+          def self.call(relationships, excluded_tables: [], included_tables: nil)
+            new(relationships, excluded_tables: excluded_tables, included_tables: included_tables).transform
+          end
+
+          def initialize(relationships, excluded_tables: [], included_tables: nil)
+            @relationships = relationships
+            @excluded_tables = Set.new(Array(excluded_tables))
+            @included_tables = included_tables ? Set.new(Array(included_tables)) : nil
+          end
+
+          def transform
+            @relationships.reject { |rel| should_filter?(rel) }
+          end
+
+          private
+
+          def should_filter?(rel)
+            return true if table_excluded?(rel[:source_table])
+            return true if table_excluded?(rel[:target_table])
+            return true if @included_tables && !table_included?(rel[:source_table])
+            return true if @included_tables && !table_included?(rel[:target_table])
+
+            false
+          end
+
+          def table_excluded?(table)
+            return false unless table
+
+            @excluded_tables.include?(table)
+          end
+
+          def table_included?(table)
+            return true unless @included_tables && table
+
+            @included_tables.include?(table)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/transformers/filter_through_relationships.rb b/lib/gitlab/reflections/relationships/transformers/filter_through_relationships.rb
new file mode 100644
index 0000000000000000000000000000000000000000..b529d2f1b3ca8796ecbd3688c2c24f739283032b
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/filter_through_relationships.rb
@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        # Filters out "through" relationships that should not be included in direct relationship mapping.
+        #
+        # Through relationships use intermediate tables to establish connections between entities
+        # and should be handled separately from direct relationships to avoid generating
+        # invalid SQL queries or incorrect relationship assumptions.
+        class FilterThroughRelationships
+          def self.call(relationships)
+            new(relationships).transform
+          end
+
+          def initialize(relationships)
+            @relationships = relationships
+          end
+
+          def transform
+            @relationships.reject { |rel| through_relationship?(rel) }
+          end
+
+          private
+
+          def through_relationship?(rel)
+            # Check if this is a through relationship using the relationship's own indicators
+            rel[:is_through_association] ||
+              (rel[:through_table] && !rel[:through_table].empty?) ||
+              (rel.respond_to?(:through_relationship?) && rel.through_relationship?)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/transformers/invert_relationships.rb b/lib/gitlab/reflections/relationships/transformers/invert_relationships.rb
new file mode 100644
index 0000000000000000000000000000000000000000..c01fbe8f486022acc22b4f5a50757e21ac18bf59
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/invert_relationships.rb
@@ -0,0 +1,76 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        class InvertRelationships
+          def self.call(relationships)
+            new(relationships).transform
+          end
+
+          def initialize(relationships)
+            @relationships = relationships
+          end
+
+          def transform
+            inverted = @relationships.dup
+
+            @relationships.each do |rel|
+              next if should_skip_inversion?(rel)
+
+              inverse_rel = create_inverse_relationship(rel)
+              inverted << inverse_rel if inverse_rel
+            end
+
+            inverted
+          end
+
+          private
+
+          def should_skip_inversion?(rel)
+            # Don't invert polymorphic base relationships (without target_table)
+            # Don't invert if it's already an inverted relationship
+            (rel[:polymorphic] && !rel[:target_table]) ||
+              rel[:inverted] == true
+          end
+
+          def create_inverse_relationship(rel)
+            return unless rel[:source_table] && rel[:target_table]
+
+            {
+              source_table: rel[:target_table],
+              target_table: rel[:source_table],
+              source_columns: rel[:target_columns],
+              target_columns: rel[:source_columns],
+              relationship_type: invert_relationship_type(rel[:relationship_type]),
+              inverted: true,
+              original_relationship: rel.slice(:source_table, :target_table, :relationship_type)
+            }.merge(preserve_polymorphic_info(rel))
+          end
+
+          def invert_relationship_type(type)
+            case type
+            when 'one_to_many' then 'many_to_one'
+            when 'many_to_one' then 'one_to_many'
+            when 'polymorphic_expanded' then 'polymorphic_expanded_inverse'
+            else type
+            end
+          end
+
+          def preserve_polymorphic_info(rel)
+            if rel[:polymorphic]
+              {
+                polymorphic: true,
+                polymorphic_type_column: rel[:polymorphic_type_column],
+                polymorphic_type_value: rel[:polymorphic_type_value]
+              }
+            else
+              {}
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/transformers/merge_data_sources.rb b/lib/gitlab/reflections/relationships/transformers/merge_data_sources.rb
new file mode 100644
index 0000000000000000000000000000000000000000..9478cf50b26ff525231a2f91b4d1a1be4ee6c0ff
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/merge_data_sources.rb
@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        class MergeDataSources
+          def self.call(relationships)
+            new(relationships).transform
+          end
+
+          def initialize(relationships)
+            @relationships = relationships
+          end
+
+          def transform
+            grouped = @relationships.group_by { |rel| key(rel) }
+            grouped.values.map { |group| merge_group(group) }
+          end
+
+          private
+
+          def key(rel)
+            [
+              rel[:source_table],
+              rel[:target_table],
+              Array(rel[:source_columns]).sort,
+              Array(rel[:target_columns]).sort
+            ]
+          end
+
+          def merge_group(group)
+            return group.first if group.size == 1
+
+            merged = group.first.dup
+            group[1..].each { |rel| merge_data(merged, rel) }
+            merged
+          end
+
+          def merge_data(base, other)
+            base[:data_sources] = Array(base[:data_sources]) | Array(other[:data_sources])
+            base[:has_ar_forward] ||= other[:has_ar_forward]
+            base[:has_ar_reverse] ||= other[:has_ar_reverse]
+            base[:has_fk_constraint] ||= other[:has_fk_constraint]
+            base[:forward_association] ||= other[:forward_association]
+            base[:reverse_association] ||= other[:reverse_association]
+            base[:fk_metadata] ||= other[:fk_metadata]
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/transformers/normalize_columns.rb b/lib/gitlab/reflections/relationships/transformers/normalize_columns.rb
new file mode 100644
index 0000000000000000000000000000000000000000..cafba872cc9ef9d2d87ec6404bb309880d12869d
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/normalize_columns.rb
@@ -0,0 +1,48 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        class NormalizeColumns
+          def self.call(relationships)
+            new(relationships).transform
+          end
+
+          def initialize(relationships)
+            @relationships = relationships
+          end
+
+          def transform
+            @relationships.map { |rel| normalize_relationship(rel) }
+          end
+
+          private
+
+          def normalize_relationship(rel)
+            normalized = rel.dup
+
+            # Set default source columns if missing
+            normalized[:source_columns] ||= ['id']
+
+            # Set target columns from foreign key if missing
+            if normalized[:foreign_key] && !normalized[:target_columns]
+              normalized[:target_columns] = [normalized[:foreign_key]]
+            end
+
+            # Ensure foreign_key_table is set
+            if normalized[:foreign_key] && !normalized[:foreign_key_table]
+              normalized[:foreign_key_table] = normalized[:target_table]
+            end
+
+            # Ensure columns are arrays
+            normalized[:source_columns] = Array(normalized[:source_columns])
+            normalized[:target_columns] = Array(normalized[:target_columns])
+
+            normalized
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/reflections/relationships/transformers/pipeline.rb b/lib/gitlab/reflections/relationships/transformers/pipeline.rb
new file mode 100644
index 0000000000000000000000000000000000000000..2be892d7763534c088e3943f52623a81afc3e1ad
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/pipeline.rb
@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        class Pipeline
+          def initialize(*transformers)
+            @transformers = transformers
+          end
+          
+          def execute(input)
+            @transformers.reduce(input) { |data, t| t.call(data) }
+          end
+        end
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/lib/gitlab/reflections/relationships/transformers/validate.rb b/lib/gitlab/reflections/relationships/transformers/validate.rb
new file mode 100644
index 0000000000000000000000000000000000000000..481b8ac95d07f51cbb171ff86b17333b03f5be39
--- /dev/null
+++ b/lib/gitlab/reflections/relationships/transformers/validate.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Reflections
+    module Relationships
+      module Transformers
+        class Validate
+          def self.call(relationships)
+            new(relationships).transform
+          end
+
+          def initialize(relationships)
+            @relationships = relationships
+          end
+
+          def transform
+            @relationships.filter_map do |rel|
+              validate_relationship(rel)
+            end
+          end
+
+          private
+
+          def validate_relationship(rel)
+            # Check required fields
+            required_fields = [:source_table, :target_table]
+            missing_fields = required_fields.select { |field| rel[field].nil? || rel[field].to_s.empty? }
+
+            if missing_fields.any?
+              Rails.logger.debug "Skipping invalid relationship: missing required fields #{missing_fields}"
+              return
+            end
+
+            # Additional validation for polymorphic relationships
+            if rel[:polymorphic] && rel[:relationship_type] == 'polymorphic_expanded' && !(rel[:polymorphic_type_column] && rel[:polymorphic_type_value])
+              Rails.logger.debug "Skipping invalid polymorphic relationship: missing type information"
+              return
+            end
+
+            # Return clean relationship hash
+            rel.compact
+          rescue StandardError => e
+            Rails.logger.debug "Failed to validate relationship: #{e.message}"
+            nil
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/tasks/database_relationships.rake b/lib/tasks/database_relationships.rake
new file mode 100644
index 0000000000000000000000000000000000000000..9ed87c6455320726f9e4c38dfbac2b30aa31a3ab
--- /dev/null
+++ b/lib/tasks/database_relationships.rake
@@ -0,0 +1,197 @@
+# frozen_string_literal: true
+
+namespace :db do
+  namespace :relationships do
+    desc "Print relationships in JSON format"
+    task all: :environment do
+      builder = Gitlab::Reflections::Relationships::Builder.new
+      relationships = builder.build_relationships
+
+      puts Gitlab::Json.pretty_generate(relationships)
+    end
+
+    desc "Crawl relationships for all tables"
+    task crawl_map: :environment do
+      builder = Gitlab::Reflections::Relationships::Builder.new
+      all_relationships = builder.build_relationships
+
+      # Filter out through associations for crawling - they don't represent direct database relationships
+      crawlable_relationships = all_relationships.reject { |rel| rel[:is_through_association] }
+
+      # Filter to only include relationships from sharded tables
+      crawlable_relationships = filter_sharded_relationships(crawlable_relationships)
+
+      # Process all relationships uniformly - polymorphic relationships are already completed
+      crawl_data = crawlable_relationships.map do |rel|
+        {
+          source_table: rel[:source_table],
+          target_table: rel[:target_table],
+          source_columns: rel[:source_columns],
+          target_columns: rel[:target_columns],
+          relationship_type: rel[:polymorphic] ? "polymorphic" : "regular",
+          # Add traceability information for debugging
+          source_info: {
+            has_ar_forward: rel[:has_ar_forward],
+            has_ar_reverse: rel[:has_ar_reverse],
+            has_fk_constraint: rel[:has_fk_constraint],
+            forward_association: rel[:forward_association]&.slice(:name, :type, :class_name),
+            reverse_association: rel[:reverse_association]&.slice(:name, :type, :class_name),
+            is_through_association: rel[:is_through_association]
+          }
+        }.tap do |item|
+          # Add polymorphic fields only if it's a polymorphic relationship
+          if rel[:polymorphic]
+            item[:polymorphic_type_column] = rel[:polymorphic_type_column]
+            # For polymorphic relationships with specific target tables, use the target table name
+            # For base polymorphic relationships (target_table: nil), keep it as nil
+            item[:polymorphic_type_value] = rel[:target_table]&.classify
+          end
+        end
+      end
+
+      # Group relationships to show duplicates and their sources
+      grouped_relationships = crawl_data.group_by do |item|
+        [
+          item[:source_table],
+          item[:target_table],
+          item[:source_columns],
+          item[:target_columns],
+          item[:relationship_type],
+          item[:polymorphic_type_column],
+          item[:polymorphic_type_value]
+        ]
+      end
+
+      # Convert grouped relationships back to crawl format
+      final_crawl_data = grouped_relationships.map do |key, group|
+        if group.size == 1
+          # Single relationship - use as is
+          group.first
+        else
+          # Multiple relationships with same key - merge source_info
+          base_relationship = group.first.dup
+          base_relationship[:source_info] = {
+            has_ar_forward: group.any? { |r| r[:source_info][:has_ar_forward] },
+            has_ar_reverse: group.any? { |r| r[:source_info][:has_ar_reverse] },
+            has_fk_constraint: group.any? { |r| r[:source_info][:has_fk_constraint] },
+            forward_association: group.find do |r|
+              r[:source_info][:forward_association]
+            end&.dig(:source_info, :forward_association),
+            reverse_association: group.find do |r|
+              r[:source_info][:reverse_association]
+            end&.dig(:source_info, :reverse_association),
+            duplicate_count: group.size,
+            all_sources: group.map { |r| r[:source_info] }
+          }
+          base_relationship
+        end
+      end
+
+      puts Gitlab::Json.pretty_generate(final_crawl_data)
+    end
+
+    desc "Crawl database relationships starting from a seed table"
+    task :crawl, [:json_file, :seed_table, :seed_id, :ignore_tables] => :environment do |_task, args|
+      unless args[:json_file] && args[:seed_table] && args[:seed_id]
+        usage_msg = "Usage: rails db:relationships:crawl[/path/to/relationships.json," \
+          "table_name,primary_key_id,table1,table2,...]"
+        puts usage_msg
+        puts "  ignore_tables: Optional comma-separated list of tables to ignore during crawling"
+        exit 1
+      end
+
+      # Parse ignore_tables if provided
+      ignore_tables = args[:ignore_tables]&.split(',')&.map(&:strip)
+
+      crawler = Gitlab::Reflections::Relationships::Crawler.new(
+        relationships_file: args[:json_file],
+        seed_table: args[:seed_table],
+        seed_id: args[:seed_id],
+        ignore_tables: ignore_tables
+      )
+
+      crawler.crawl
+    end
+
+    desc "Print polymorphic relationships using predicates in JSON format"
+    task polymorphic_json: :environment do
+      puts "Building relationships..."
+      builder = Gitlab::Reflections::Relationships::Builder.new
+      relationships = builder.build_relationships
+
+      puts "Filtering polymorphic relationships using predicates..."
+      # Work directly with Array - no wrapper needed
+      polymorphic_rels = relationships.select { |rel| rel[:polymorphic] == true }
+
+      # Group by polymorphic key and aggregate
+      grouped = polymorphic_rels.group_by do |rel|
+        next unless rel[:polymorphic] == true
+
+        [
+          rel[:polymorphic_type_column],
+          rel[:foreign_key],
+          rel[:foreign_key_table] || rel[:target_table]
+        ]
+      end
+
+      puts "Aggregating #{grouped.size} polymorphic relationship groups..."
+      aggregated_polymorphic = grouped.map do |key, group|
+        type_column, foreign_key, foreign_key_table = key
+
+        # Collect target tables from the group (polymorphic relationships may not have target_table)
+        targets = group.filter_map { |rel| rel[:target_table] }.compact.uniq.sort
+
+        # For polymorphic relationships, we need to discover targets by analyzing related relationships
+        # that share the same polymorphic_type_column
+        if targets.empty? && type_column
+          # Find other relationships that might be targets for this polymorphic association
+          potential_targets = relationships
+            .select do |rel|
+              rel[:polymorphic_type_column] == type_column &&
+                rel[:has_ar_forward] &&
+                rel[:source_table] != foreign_key_table
+            end
+            .filter_map { |rel| rel[:source_table] }
+            .compact.uniq.sort
+
+          targets = potential_targets
+        end
+
+        {
+          polymorphic_type_column: type_column,
+          foreign_key: foreign_key,
+          foreign_key_table: foreign_key_table,
+          target_tables: targets,
+          has_ar_forward: group.any? { |rel| rel[:has_ar_forward] },
+          has_ar_reverse: group.any? { |rel| rel[:has_ar_reverse] },
+          has_fk_constraint: group.any? { |rel| rel[:has_fk_constraint] },
+          relationship_count: group.size,
+          sample_relationship: group.first # Include a sample for debugging
+        }
+      end
+
+      output = aggregated_polymorphic.sort_by { |rel| rel[:polymorphic_type_column] || '' }
+
+      puts "\n" + ("=" * 80)
+      puts "POLYMORPHIC RELATIONSHIPS (using predicates)"
+      puts "=" * 80
+      puts Gitlab::Json.pretty_generate(output)
+    end
+
+    private
+
+    def filter_sharded_relationships(relationships)
+      # Get all sharded table names
+      sharded_tables = Set.new
+
+      Gitlab::Database::Dictionary.entries.each do |entry|
+        sharded_tables.add(entry.table_name) if entry.sharding_key.present?
+      end
+
+      # Filter relationships to only include those from sharded tables
+      relationships.select do |rel|
+        sharded_tables.include?(rel[:source_table])
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/database/foreign_keys_spec.rb b/spec/lib/gitlab/reflections/database/foreign_keys_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..4f1b42f734cf9396f98e4bc3eac46419b57629f2
--- /dev/null
+++ b/spec/lib/gitlab/reflections/database/foreign_keys_spec.rb
@@ -0,0 +1,255 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Database::ForeignKeys, feature_category: :database do
+  # Mock PostgresForeignKey class to avoid database dependencies
+  let(:mock_postgres_foreign_key_class) do
+    Class.new do
+      def self.not_inherited
+        MockScope.new
+      end
+    end
+  end
+
+  # Mock scope class for chaining ActiveRecord-like queries
+  class MockScope
+    def initialize(filtered_tables: [], filtered_patterns: [])
+      @filtered_tables = filtered_tables
+      @filtered_patterns = filtered_patterns
+    end
+
+    def where
+      MockWhereClause.new(self)
+    end
+
+    def to_a
+      # Return mock foreign key records that would pass the filters
+      [
+        mock_foreign_key('users', 'projects', 'user_id'),
+        mock_foreign_key('issues', 'projects', 'project_id'),
+        mock_foreign_key('merge_requests', 'projects', 'target_project_id')
+      ].reject do |fk|
+        @filtered_tables.include?(fk.constrained_table_name) ||
+          @filtered_tables.include?(fk.referenced_table_name) ||
+          @filtered_patterns.any? { |pattern| fk.constrained_table_name.match?(pattern) } ||
+          @filtered_patterns.any? { |pattern| fk.referenced_table_name.match?(pattern) }
+      end
+    end
+
+    private
+
+    def mock_foreign_key(constrained_table, referenced_table, column)
+      double('foreign_key',
+        constrained_table_name: constrained_table,
+        referenced_table_name: referenced_table,
+        constrained_column_names: [column]
+      )
+    end
+  end
+
+  # Mock where clause for building query conditions
+  class MockWhereClause
+    def initialize(scope)
+      @scope = scope
+    end
+
+    def not(conditions)
+      if conditions.is_a?(Hash)
+        if conditions.key?(:constrained_table_name)
+          @scope.instance_variable_get(:@filtered_tables) << conditions[:constrained_table_name]
+        elsif conditions.key?(:referenced_table_name)
+          @scope.instance_variable_get(:@filtered_tables) << conditions[:referenced_table_name]
+        end
+      elsif conditions.is_a?(String) && conditions.include?('~')
+        # Handle pattern matching conditions
+        pattern_match = conditions.match(/(\w+_table_name) ~ \?/)
+        if pattern_match
+          # This would be handled by the calling code passing the pattern
+          # We'll track this in the test
+        end
+      end
+      @scope
+    end
+  end
+
+  before do
+    # Stub the PostgresForeignKey constant
+    stub_const('Gitlab::Database::PostgresForeignKey', mock_postgres_foreign_key_class)
+  end
+
+  describe '.for_relationships' do
+    subject { described_class.for_relationships }
+
+    it 'returns a scope that filters out internal tables' do
+      expect(mock_postgres_foreign_key_class).to receive(:not_inherited).and_return(MockScope.new)
+      
+      result = subject
+      expect(result).to be_a(MockScope)
+    end
+
+    it 'filters out schema_migrations table' do
+      scope = MockScope.new
+      allow(mock_postgres_foreign_key_class).to receive(:not_inherited).and_return(scope)
+      
+      # Mock the where.not chaining for schema_migrations
+      expect(scope).to receive(:where).and_return(double('where_clause', not: scope)).at_least(:once)
+      
+      described_class.for_relationships
+    end
+
+    it 'filters out ar_internal_metadata table' do
+      scope = MockScope.new
+      allow(mock_postgres_foreign_key_class).to receive(:not_inherited).and_return(scope)
+      
+      # Mock the where.not chaining for ar_internal_metadata
+      expect(scope).to receive(:where).and_return(double('where_clause', not: scope)).at_least(:once)
+      
+      described_class.for_relationships
+    end
+
+    context 'with system table patterns' do
+      it 'filters out batched_background_migration tables' do
+        scope = MockScope.new
+        allow(mock_postgres_foreign_key_class).to receive(:not_inherited).and_return(scope)
+        
+        # Mock the pattern-based filtering
+        expect(scope).to receive(:where).and_return(double('where_clause', not: scope)).at_least(:once)
+        
+        described_class.for_relationships
+      end
+
+      it 'filters out postgres_ prefixed tables' do
+        scope = MockScope.new
+        allow(mock_postgres_foreign_key_class).to receive(:not_inherited).and_return(scope)
+        
+        expect(scope).to receive(:where).and_return(double('where_clause', not: scope)).at_least(:once)
+        
+        described_class.for_relationships
+      end
+
+      it 'filters out detached_partitions table' do
+        scope = MockScope.new
+        allow(mock_postgres_foreign_key_class).to receive(:not_inherited).and_return(scope)
+        
+        expect(scope).to receive(:where).and_return(double('where_clause', not: scope)).at_least(:once)
+        
+        described_class.for_relationships
+      end
+
+      it 'filters out loose_foreign_keys_deleted_records table' do
+        scope = MockScope.new
+        allow(mock_postgres_foreign_key_class).to receive(:not_inherited).and_return(scope)
+        
+        expect(scope).to receive(:where).and_return(double('where_clause', not: scope)).at_least(:once)
+        
+        described_class.for_relationships
+      end
+
+      it 'filters out ghost_user_migrations table' do
+        scope = MockScope.new
+        allow(mock_postgres_foreign_key_class).to receive(:not_inherited).and_return(scope)
+        
+        expect(scope).to receive(:where).and_return(double('where_clause', not: scope)).at_least(:once)
+        
+        described_class.for_relationships
+      end
+    end
+
+    context 'with realistic foreign key data' do
+      let(:mock_foreign_keys) do
+        [
+          # Valid application foreign keys
+          double('fk1', constrained_table_name: 'issues', referenced_table_name: 'projects'),
+          double('fk2', constrained_table_name: 'merge_requests', referenced_table_name: 'users'),
+          
+          # Internal tables that should be filtered out
+          double('fk3', constrained_table_name: 'schema_migrations', referenced_table_name: 'projects'),
+          double('fk4', constrained_table_name: 'issues', referenced_table_name: 'ar_internal_metadata'),
+          
+          # System tables that should be filtered out
+          double('fk5', constrained_table_name: 'batched_background_migration_jobs', referenced_table_name: 'projects'),
+          double('fk6', constrained_table_name: 'postgres_indexes', referenced_table_name: 'users'),
+          double('fk7', constrained_table_name: 'detached_partitions', referenced_table_name: 'projects'),
+          double('fk8', constrained_table_name: 'loose_foreign_keys_deleted_records', referenced_table_name: 'users'),
+          double('fk9', constrained_table_name: 'ghost_user_migrations', referenced_table_name: 'projects')
+        ]
+      end
+
+      it 'includes valid application foreign keys' do
+        # This test demonstrates the expected behavior without actually executing database queries
+        valid_tables = %w[issues merge_requests projects users]
+        invalid_tables = described_class::INTERNAL_TABLE_NAMES
+        invalid_patterns = described_class::SYSTEM_TABLE_PATTERNS
+
+        mock_foreign_keys.each do |fk|
+          should_be_included = !invalid_tables.include?(fk.constrained_table_name) &&
+                              !invalid_tables.include?(fk.referenced_table_name) &&
+                              !invalid_patterns.any? { |pattern| fk.constrained_table_name.match?(pattern) } &&
+                              !invalid_patterns.any? { |pattern| fk.referenced_table_name.match?(pattern) }
+
+          if should_be_included
+            expect(valid_tables).to include(fk.constrained_table_name)
+            expect(valid_tables).to include(fk.referenced_table_name)
+          end
+        end
+      end
+    end
+  end
+
+  describe 'constants' do
+    describe 'INTERNAL_TABLE_NAMES' do
+      it 'includes schema_migrations' do
+        expect(described_class::INTERNAL_TABLE_NAMES).to include('schema_migrations')
+      end
+
+      it 'includes ar_internal_metadata' do
+        expect(described_class::INTERNAL_TABLE_NAMES).to include('ar_internal_metadata')
+      end
+
+      it 'contains only string values' do
+        described_class::INTERNAL_TABLE_NAMES.each do |table_name|
+          expect(table_name).to be_a(String)
+        end
+      end
+    end
+
+    describe 'SYSTEM_TABLE_PATTERNS' do
+      it 'includes batched_background_migration pattern' do
+        pattern = described_class::SYSTEM_TABLE_PATTERNS.find { |p| p.source.include?('batched_background_migration') }
+        expect(pattern).not_to be_nil
+        expect('batched_background_migration_jobs').to match(pattern)
+      end
+
+      it 'includes postgres_ pattern' do
+        pattern = described_class::SYSTEM_TABLE_PATTERNS.find { |p| p.source.include?('postgres_') }
+        expect(pattern).not_to be_nil
+        expect('postgres_indexes').to match(pattern)
+      end
+
+      it 'includes detached_partitions pattern' do
+        pattern = described_class::SYSTEM_TABLE_PATTERNS.find { |p| p.source.include?('detached_partitions') }
+        expect(pattern).not_to be_nil
+        expect('detached_partitions').to match(pattern)
+      end
+
+      it 'includes loose_foreign_keys_deleted_records pattern' do
+        pattern = described_class::SYSTEM_TABLE_PATTERNS.find { |p| p.source.include?('loose_foreign_keys_deleted_records') }
+        expect(pattern).not_to be_nil
+        expect('loose_foreign_keys_deleted_records').to match(pattern)
+      end
+
+      it 'includes ghost_user_migrations pattern' do
+        pattern = described_class::SYSTEM_TABLE_PATTERNS.find { |p| p.source.include?('ghost_user_migrations') }
+        expect(pattern).not_to be_nil
+        expect('ghost_user_migrations').to match(pattern)
+      end
+
+      it 'contains only Regexp objects' do
+        described_class::SYSTEM_TABLE_PATTERNS.each do |pattern|
+          expect(pattern).to be_a(Regexp)
+        end
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/lib/gitlab/reflections/models/active_record_spec.rb b/spec/lib/gitlab/reflections/models/active_record_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..006f641da8dcdedf026491411d0184556cd70a11
--- /dev/null
+++ b/spec/lib/gitlab/reflections/models/active_record_spec.rb
@@ -0,0 +1,452 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Models::ActiveRecord, feature_category: :database do
+  # Mock ActiveRecord::Base to avoid loading actual models
+  let(:mock_active_record_base) do
+    Class.new do
+      def self.descendants
+        [
+          MockUserModel,
+          MockProjectModel,
+          MockIssueModel,
+          MockAbstractModel,
+          MockAnonymousModel,
+          MockActiveRecordInternalModel,
+          MockActiveStorageModel,
+          MockTestModel,
+          MockFixtureModel,
+          MockRSpecModel
+        ]
+      end
+    end
+  end
+
+  # Mock application models
+  class MockUserModel
+    def self.name
+      'User'
+    end
+
+    def self.table_name
+      'users'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  class MockProjectModel
+    def self.name
+      'Project'
+    end
+
+    def self.table_name
+      'projects'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  class MockIssueModel
+    def self.name
+      'Issue'
+    end
+
+    def self.table_name
+      'issues'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  # Mock models that should be excluded
+  class MockAbstractModel
+    def self.name
+      'AbstractModel'
+    end
+
+    def self.table_name
+      'abstract_models'
+    end
+
+    def self.abstract_class?
+      true
+    end
+  end
+
+  class MockAnonymousModel
+    def self.name
+      nil
+    end
+
+    def self.table_name
+      'anonymous_table'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  class MockActiveRecordInternalModel
+    def self.name
+      'ActiveRecord::InternalModel'
+    end
+
+    def self.table_name
+      'internal_table'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  class MockActiveStorageModel
+    def self.name
+      'ActiveStorage::Blob'
+    end
+
+    def self.table_name
+      'active_storage_blobs'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  class MockTestModel
+    def self.name
+      'UserTest'
+    end
+
+    def self.table_name
+      'user_tests'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  class MockFixtureModel
+    def self.name
+      'ProjectFixture'
+    end
+
+    def self.table_name
+      'project_fixtures'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  class MockRSpecModel
+    def self.name
+      'RSpec::TestModel'
+    end
+
+    def self.table_name
+      'rspec_test_models'
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  class MockModelWithoutTable
+    def self.name
+      'ModelWithoutTable'
+    end
+
+    def self.table_name
+      nil
+    end
+
+    def self.abstract_class?
+      false
+    end
+  end
+
+  before do
+    # Stub ActiveRecord::Base constant
+    stub_const('ActiveRecord::Base', mock_active_record_base)
+    
+    # Clear any existing singleton instance
+    described_class.instance_variable_set(:@singleton__instance__, nil)
+  end
+
+  describe '#initialize' do
+    it 'builds indexes on initialization' do
+      instance = described_class.instance
+      
+      expect(instance.table_to_model_index).to be_a(Hash)
+      expect(instance.model_to_table_index).to be_a(Hash)
+      expect(instance.models).to be_an(Array)
+    end
+  end
+
+  describe '#models' do
+    subject { described_class.instance.models }
+
+    it 'includes valid application models' do
+      expect(subject).to include(MockUserModel)
+      expect(subject).to include(MockProjectModel)
+      expect(subject).to include(MockIssueModel)
+    end
+
+    it 'excludes abstract models' do
+      expect(subject).not_to include(MockAbstractModel)
+    end
+
+    it 'excludes anonymous models' do
+      expect(subject).not_to include(MockAnonymousModel)
+    end
+
+    it 'excludes ActiveRecord internal models' do
+      expect(subject).not_to include(MockActiveRecordInternalModel)
+    end
+
+    it 'excludes ActiveStorage models' do
+      expect(subject).not_to include(MockActiveStorageModel)
+    end
+
+    it 'excludes test models' do
+      expect(subject).not_to include(MockTestModel)
+    end
+
+    it 'excludes fixture models' do
+      expect(subject).not_to include(MockFixtureModel)
+    end
+
+    it 'excludes RSpec models' do
+      expect(subject).not_to include(MockRSpecModel)
+    end
+
+    it 'excludes models without table names' do
+      # Add a model without table name to the descendants
+      allow(mock_active_record_base).to receive(:descendants).and_return([
+        MockUserModel,
+        MockProjectModel,
+        MockModelWithoutTable
+      ])
+      
+      # Clear singleton to force rebuild
+      described_class.instance_variable_set(:@singleton__instance__, nil)
+      
+      expect(described_class.instance.models).not_to include(MockModelWithoutTable)
+    end
+
+    it 'handles duplicate model names' do
+      # Create duplicate models with same name
+      duplicate_model = Class.new do
+        def self.name
+          'User'
+        end
+
+        def self.table_name
+          'duplicate_users'
+        end
+
+        def self.abstract_class?
+          false
+        end
+      end
+
+      allow(mock_active_record_base).to receive(:descendants).and_return([
+        MockUserModel,
+        duplicate_model
+      ])
+      
+      # Clear singleton to force rebuild
+      described_class.instance_variable_set(:@singleton__instance__, nil)
+      
+      # Should only include one of the duplicate models
+      user_models = described_class.instance.models.select { |m| m.name == 'User' }
+      expect(user_models.length).to eq(1)
+    end
+  end
+
+  describe '#table_to_model_index' do
+    subject { described_class.instance.table_to_model_index }
+
+    it 'maps table names to model names' do
+      expect(subject['users']).to eq('User')
+      expect(subject['projects']).to eq('Project')
+      expect(subject['issues']).to eq('Issue')
+    end
+
+    it 'does not include excluded models' do
+      expect(subject).not_to have_key('abstract_models')
+      expect(subject).not_to have_key('internal_table')
+      expect(subject).not_to have_key('active_storage_blobs')
+      expect(subject).not_to have_key('user_tests')
+      expect(subject).not_to have_key('project_fixtures')
+      expect(subject).not_to have_key('rspec_test_models')
+    end
+
+    it 'does not include anonymous models' do
+      expect(subject).not_to have_key('anonymous_table')
+    end
+  end
+
+  describe '#model_to_table_index' do
+    subject { described_class.instance.model_to_table_index }
+
+    it 'maps model names to table names' do
+      expect(subject['User']).to eq('users')
+      expect(subject['Project']).to eq('projects')
+      expect(subject['Issue']).to eq('issues')
+    end
+
+    it 'does not include excluded models' do
+      expect(subject).not_to have_key('AbstractModel')
+      expect(subject).not_to have_key('ActiveRecord::InternalModel')
+      expect(subject).not_to have_key('ActiveStorage::Blob')
+      expect(subject).not_to have_key('UserTest')
+      expect(subject).not_to have_key('ProjectFixture')
+      expect(subject).not_to have_key('RSpec::TestModel')
+    end
+  end
+
+  describe 'singleton behavior' do
+    it 'returns the same instance on multiple calls' do
+      instance1 = described_class.instance
+      instance2 = described_class.instance
+      
+      expect(instance1).to be(instance2)
+    end
+
+    it 'builds indexes only once' do
+      expect(mock_active_record_base).to receive(:descendants).once.and_call_original
+      
+      described_class.instance
+      described_class.instance
+    end
+  end
+
+  describe 'private methods' do
+    let(:instance) { described_class.instance }
+
+    describe '#discover_models' do
+      it 'returns ActiveRecord::Base descendants' do
+        models = instance.send(:discover_models)
+        expect(models).to include(MockUserModel)
+        expect(models).to include(MockProjectModel)
+      end
+    end
+
+    describe '#skip_model?' do
+      it 'skips abstract models' do
+        expect(instance.send(:skip_model?, MockAbstractModel)).to be true
+      end
+
+      it 'skips anonymous models' do
+        expect(instance.send(:skip_model?, MockAnonymousModel)).to be true
+      end
+
+      it 'skips models without table names' do
+        expect(instance.send(:skip_model?, MockModelWithoutTable)).to be true
+      end
+
+      it 'skips excluded models' do
+        expect(instance.send(:skip_model?, MockActiveRecordInternalModel)).to be true
+        expect(instance.send(:skip_model?, MockActiveStorageModel)).to be true
+        expect(instance.send(:skip_model?, MockTestModel)).to be true
+        expect(instance.send(:skip_model?, MockFixtureModel)).to be true
+        expect(instance.send(:skip_model?, MockRSpecModel)).to be true
+      end
+
+      it 'does not skip valid application models' do
+        expect(instance.send(:skip_model?, MockUserModel)).to be false
+        expect(instance.send(:skip_model?, MockProjectModel)).to be false
+        expect(instance.send(:skip_model?, MockIssueModel)).to be false
+      end
+    end
+
+    describe '#excluded_model?' do
+      it 'excludes ActiveRecord internal models' do
+        expect(instance.send(:excluded_model?, MockActiveRecordInternalModel)).to be true
+      end
+
+      it 'excludes ActiveStorage models' do
+        expect(instance.send(:excluded_model?, MockActiveStorageModel)).to be true
+      end
+
+      it 'excludes test models' do
+        expect(instance.send(:excluded_model?, MockTestModel)).to be true
+      end
+
+      it 'excludes fixture models' do
+        expect(instance.send(:excluded_model?, MockFixtureModel)).to be true
+      end
+
+      it 'excludes RSpec models' do
+        expect(instance.send(:excluded_model?, MockRSpecModel)).to be true
+      end
+
+      it 'does not exclude application models' do
+        expect(instance.send(:excluded_model?, MockUserModel)).to be false
+        expect(instance.send(:excluded_model?, MockProjectModel)).to be false
+      end
+    end
+
+    describe '#filter_models' do
+      let(:all_models) do
+        [
+          MockUserModel,
+          MockProjectModel,
+          MockAbstractModel,
+          MockAnonymousModel,
+          MockActiveRecordInternalModel,
+          MockTestModel
+        ]
+      end
+
+      it 'filters out excluded models' do
+        filtered = instance.send(:filter_models, all_models)
+        
+        expect(filtered).to include(MockUserModel)
+        expect(filtered).to include(MockProjectModel)
+        expect(filtered).not_to include(MockAbstractModel)
+        expect(filtered).not_to include(MockAnonymousModel)
+        expect(filtered).not_to include(MockActiveRecordInternalModel)
+        expect(filtered).not_to include(MockTestModel)
+      end
+
+      it 'handles duplicate model names' do
+        duplicate_user = Class.new do
+          def self.name
+            'User'
+          end
+
+          def self.table_name
+            'users'
+          end
+
+          def self.abstract_class?
+            false
+          end
+        end
+
+        models_with_duplicate = [MockUserModel, duplicate_user]
+        filtered = instance.send(:filter_models, models_with_duplicate)
+        
+        # Should only include one User model
+        user_models = filtered.select { |m| m.name == 'User' }
+        expect(user_models.length).to eq(1)
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/lib/gitlab/reflections/polymorphic_edge_cases_spec.rb b/spec/lib/gitlab/reflections/polymorphic_edge_cases_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..aaeaa700babad9ee5a3fc0a4608922584908e2a2
--- /dev/null
+++ b/spec/lib/gitlab/reflections/polymorphic_edge_cases_spec.rb
@@ -0,0 +1,215 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe 'Polymorphic Relationship Edge Cases', feature_category: :database do
+  describe 'complex polymorphic scenarios' do
+    context 'with multiple polymorphic targets' do
+      it 'handles models that can be targets of multiple polymorphic relationships' do
+        # Test that models like User can be targets of multiple polymorphic relationships
+        # e.g., author_type, assignee_type, etc.
+
+        builder = Gitlab::Reflections::Relationships::Builder.new
+        relationships = builder.build_relationships
+
+        # Find all relationships where User is a polymorphic target
+        user_polymorphic_relationships = relationships.select do |relationship|
+          relationship[:polymorphic] &&
+            relationship[:polymorphic_targets]&.include?('User')
+        end
+
+        expect(user_polymorphic_relationships).not_to be_empty
+
+        # Verify that each relationship has proper polymorphic metadata
+        user_polymorphic_relationships.each do |relationship|
+          expect(relationship[:polymorphic_type_column]).to be_present
+          expect(relationship[:polymorphic_name]).to be_present
+          expect(relationship[:polymorphic_targets]).to include('User')
+        end
+      end
+    end
+
+    context 'with polymorphic relationships without matching counterparts' do
+      it 'handles belongs_to polymorphic without corresponding has_many' do
+        # Create a test model with orphaned polymorphic belongs_to
+        test_model = Class.new(ApplicationRecord) do
+          self.table_name = 'test_orphaned_poly'
+          belongs_to :orphaned_target, polymorphic: true
+        end
+
+        # Mock the reflection
+        reflection = double('reflection',
+          macro: :belongs_to,
+          polymorphic?: true,
+          foreign_key: 'orphaned_target_id',
+          foreign_type: 'orphaned_target_type',
+          options: {}
+        )
+
+        handler = Gitlab::Reflections::Relationships::Handlers::BelongsToHandler.new(
+          test_model, :orphaned_target, reflection
+        )
+
+        relationship = handler.build_relationship
+
+        expect(relationship[:polymorphic]).to be true
+        expect(relationship[:polymorphic_type_column]).to eq('orphaned_target_type')
+        expect(relationship[:polymorphic_targets]).to eq([])
+      end
+    end
+
+    context 'with custom polymorphic column names' do
+      it 'handles custom foreign_type column names' do
+        # Test polymorphic relationships with custom type column names
+        test_model = Class.new(ApplicationRecord) do
+          self.table_name = 'test_custom_poly'
+          belongs_to :custom_target, polymorphic: true, foreign_type: 'custom_type_col'
+        end
+
+        reflection = double('reflection',
+          macro: :belongs_to,
+          polymorphic?: true,
+          foreign_key: 'custom_target_id',
+          foreign_type: 'custom_type_col',
+          options: { foreign_type: 'custom_type_col' }
+        )
+
+        handler = Gitlab::Reflections::Relationships::Handlers::BelongsToHandler.new(
+          test_model, :custom_target, reflection
+        )
+
+        relationship = handler.build_relationship
+
+        expect(relationship[:polymorphic_type_column]).to eq('custom_type_col')
+      end
+    end
+
+    context 'with nested polymorphic relationships' do
+      it 'handles polymorphic relationships in STI hierarchies' do
+        # Test polymorphic relationships where the target is part of STI hierarchy
+        builder = Gitlab::Reflections::Relationships::Builder.new
+        relationships = builder.build_relationships
+
+        # Find polymorphic relationships that involve STI models
+        sti_polymorphic_relationships = relationships.select do |relationship|
+          relationship[:polymorphic] &&
+            relationship[:polymorphic_targets]&.any? { |target| target.include?('::') }
+        end
+
+        # Verify STI models are handled correctly
+        sti_polymorphic_relationships.each do |relationship|
+          expect(relationship[:polymorphic_targets]).to be_an(Array)
+          expect(relationship[:polymorphic_type_column]).to be_present
+        end
+      end
+    end
+
+    context 'with polymorphic through relationships' do
+      it 'handles has_many through with polymorphic associations' do
+        # Test complex relationships like has_many :something, through: :polymorphic_association
+        builder = Gitlab::Reflections::Relationships::Builder.new
+        relationships = builder.build_relationships
+
+        # Find through relationships that involve polymorphic associations
+        through_polymorphic_edges = relationships.select do |edge|
+          edge[:through_table] && edge[:polymorphic]
+        end
+
+        through_polymorphic_edges.each do |edge|
+          expect(edge[:through_table]).to be_present
+          expect(edge[:polymorphic]).to be true
+          expect(edge[:polymorphic_type_column]).to be_present
+        end
+      end
+    end
+
+    context 'with bidirectional polymorphic relationships' do
+      it 'correctly merges polymorphic belongs_to with has_many as: relationships' do
+        builder = Gitlab::Reflections::Relationships::Builder.new
+        relationships = builder.build_relationships
+
+        # Find bidirectional polymorphic relationships
+        bidirectional_polymorphic = relationships.select do |edge|
+          edge[:polymorphic] &&
+            edge[:has_ar_forward] &&
+            edge[:has_ar_reverse]
+        end
+
+        bidirectional_polymorphic.each do |edge|
+          expect(edge[:forward_association]).to be_present
+          expect(edge[:reverse_association]).to be_present
+          expect(edge[:polymorphic_type_column]).to be_present
+          expect(edge[:polymorphic_targets]).to be_an(Array)
+        end
+      end
+    end
+
+    context 'with Member model edge cases' do
+      it 'handles Member polymorphic relationship correctly' do
+        builder = Gitlab::Reflections::Relationships::Builder.new
+        relationships = builder.build_relationships
+
+        # Find Member's polymorphic relationship
+        member_edges = relationships.select do |edge|
+          edge[:source_table] == 'members' || edge[:target_table] == 'members'
+        end
+
+        # Should have polymorphic edges for Member
+        polymorphic_member_edges = member_edges.select { |edge| edge[:polymorphic] }
+        expect(polymorphic_member_edges).not_to be_empty
+
+        polymorphic_member_edges.each do |edge|
+          if edge[:source_table] == 'members'
+            # belongs_to :source, polymorphic: true
+            expect(edge[:polymorphic_type_column]).to eq('source_type')
+            expect(edge[:has_ar_reverse]).to be true
+          else
+            # has_many :members, as: :source (from Group/Project)
+            expect(edge[:polymorphic_type_column]).to eq('source_type')
+            expect(edge[:has_ar_forward]).to be true
+          end
+        end
+      end
+
+      it 'identifies Group and Project as Member polymorphic targets' do
+        builder = Gitlab::Reflections::Relationships::Builder.new
+        relationships = builder.build_relationships
+
+        # Find the specific polymorphic relationship for Member's source
+        member_source_edges = relationships.select do |edge|
+          edge[:polymorphic] &&
+            edge[:polymorphic_type_column] == 'source_type' &&
+            (edge[:source_table] == 'members' || edge[:target_table] == 'members')
+        end
+
+        expect(member_source_edges).not_to be_empty
+
+        # Collect all polymorphic targets from these edges
+        all_targets = member_source_edges.flat_map { |edge| edge[:polymorphic_targets] || [] }.uniq
+
+        expect(all_targets).to include('Group')
+        expect(all_targets).to include('Project')
+      end
+    end
+  end
+
+  describe 'performance considerations' do
+    it 'handles large numbers of polymorphic relationships efficiently' do
+      # Test that the system can handle many polymorphic relationships without performance issues
+      start_time = Time.current
+
+      builder = Gitlab::Reflections::Relationships::Builder.new
+      relationships = builder.build_relationships
+
+      end_time = Time.current
+      duration = end_time - start_time
+
+      # Should complete within reasonable time (adjust threshold as needed)
+      expect(duration).to be < 30.seconds
+
+      # Should find polymorphic relationships
+      polymorphic_edges = relationships.select { |edge| edge[:polymorphic] }
+      expect(polymorphic_edges).not_to be_empty
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/polymorphic_json_task_spec.rb b/spec/lib/gitlab/reflections/polymorphic_json_task_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..8458b9062b520e8ee522e9304bf2dc480bc5e67a
--- /dev/null
+++ b/spec/lib/gitlab/reflections/polymorphic_json_task_spec.rb
@@ -0,0 +1,167 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+require 'rake'
+
+RSpec.describe 'rails db:relationships:polymorphic_json task', feature_category: :database do
+  before_all do
+    Rake.application.rake_require 'tasks/database_relationships'
+  end
+
+  let(:task) { Rake::Task['db:relationships:polymorphic_json'] }
+
+  before do
+    task.reenable
+  end
+
+  describe 'polymorphic_json task output' do
+    it 'includes Member model relationships in the output' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      # Member model should appear in the output with source_type column
+      expect(parsed_output).to have_key('members')
+      expect(parsed_output['members']).to have_key('source_type')
+      expect(parsed_output['members']['source_type']).to include('Group', 'Project')
+    end
+
+    it 'includes all polymorphic relationships from Group model' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      # Groups should show up as targets for members polymorphic relationship
+      expect(parsed_output['members']['source_type']).to include('Group')
+    end
+
+    it 'includes all polymorphic relationships from Project model' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      # Projects should show up as targets for members polymorphic relationship
+      expect(parsed_output['members']['source_type']).to include('Project')
+    end
+
+    it 'correctly identifies polymorphic type columns' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      # Verify that common polymorphic type columns are present
+      expect(parsed_output).to have_key('notes')
+      expect(parsed_output['notes']).to have_key('noteable_type')
+
+      expect(parsed_output).to have_key('todos')
+      expect(parsed_output['todos']).to have_key('target_type')
+    end
+
+    it 'handles multiple polymorphic relationships on the same table' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      # Some tables have multiple polymorphic relationships
+      # For example, events table has both target_type and author_type
+      expect(parsed_output['events'].keys.length).to be >= 1 if parsed_output.key?('events')
+    end
+
+    it 'excludes non-polymorphic relationships' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      # Regular foreign key relationships should not appear
+      # Only polymorphic type columns should be present
+      parsed_output.each do |table_name, columns|
+        columns.each do |column_name, models|
+          expect(column_name).to end_with('_type'),
+            "Expected #{table_name}.#{column_name} to be a polymorphic type column"
+          expect(models).to be_an(Array),
+            "Expected #{table_name}.#{column_name} to have an array of model names"
+        end
+      end
+    end
+
+    it 'sorts model names alphabetically within each polymorphic relationship' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      parsed_output.each do |table_name, columns|
+        columns.each do |column_name, models|
+          expect(models).to eq(models.sort),
+            "Expected #{table_name}.#{column_name} models to be sorted alphabetically"
+        end
+      end
+    end
+
+    context 'with Member model specifically' do
+      it 'correctly identifies Group and Project as polymorphic targets' do
+        output = capture_task_output { task.invoke }
+        parsed_output = Gitlab::Json.parse(output)
+
+        expect(parsed_output).to have_key('members')
+        expect(parsed_output['members']).to have_key('source_type')
+
+        member_targets = parsed_output['members']['source_type']
+        expect(member_targets).to include('Group')
+        expect(member_targets).to include('Project')
+        expect(member_targets).to be_sorted
+      end
+
+      it 'does not include other models as Member polymorphic targets' do
+        output = capture_task_output { task.invoke }
+        parsed_output = Gitlab::Json.parse(output)
+
+        member_targets = parsed_output['members']['source_type']
+
+        # Member's source should only be Group and Project
+        # It should not include other models like User, Issue, etc.
+        expect(member_targets).to match_array(%w[Group Project])
+      end
+    end
+  end
+
+  describe 'edge cases' do
+    it 'handles empty polymorphic relationships gracefully' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      # Some polymorphic relationships might have empty arrays
+      # This should be handled gracefully
+      parsed_output.each do |table_name, columns|
+        columns.each do |column_name, models|
+          expect(models).to be_an(Array)
+          # Empty arrays are acceptable
+        end
+      end
+    end
+
+    it 'produces valid JSON output' do
+      output = capture_task_output { task.invoke }
+
+      expect { Gitlab::Json.parse(output) }.not_to raise_error
+    end
+
+    it 'handles models with multiple polymorphic associations' do
+      output = capture_task_output { task.invoke }
+      parsed_output = Gitlab::Json.parse(output)
+
+      # Some models might have multiple polymorphic associations
+      # Each should be handled correctly
+      parsed_output.each do |table_name, columns|
+        expect(columns).to be_a(Hash)
+        columns.each do |column_name, models|
+          expect(column_name).to be_a(String)
+          expect(models).to be_an(Array)
+        end
+      end
+    end
+  end
+
+  private
+
+  def capture_task_output
+    original_stdout = $stdout
+    $stdout = StringIO.new
+    yield
+    $stdout.string
+  ensure
+    $stdout = original_stdout
+  end
+end
diff --git a/spec/lib/gitlab/reflections/polymorphic_relationships_spec.rb b/spec/lib/gitlab/reflections/polymorphic_relationships_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..4369948185b1239c21d4b1e3172d198393bae085
--- /dev/null
+++ b/spec/lib/gitlab/reflections/polymorphic_relationships_spec.rb
@@ -0,0 +1,159 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe 'Member polymorphic relationships in rails task', feature_category: :database do
+  describe 'polymorphic_json task logic' do
+    it 'correctly processes Member polymorphic relationships' do
+      # This test documents the expected behavior for Member polymorphic relationships
+      # in the rails db:relationships:polymorphic_json task
+
+      # Mock edges representing Member's polymorphic relationship structure
+      member_belongs_to_edge = {
+        source_table: 'members',
+        target_table: nil,
+        polymorphic: true,
+        polymorphic_type_column: 'source_type',
+        has_ar_reverse: true,
+        reverse_association: {
+          name: 'source',
+          type: 'belongs_to',
+          model: 'GroupMember'
+        }
+      }
+
+      # Mock Group's has_many relationship with as: :source
+      namespace_has_many_edge = {
+        source_table: 'namespaces',
+        target_table: 'members',
+        polymorphic: true,
+        polymorphic_type_column: 'source_type',
+        has_ar_forward: true,
+        forward_association: {
+          name: 'group_members',
+          type: 'has_many',
+          model: 'Group'
+        }
+      }
+
+      # Mock Project's has_many relationship with as: :source
+      project_has_many_edge = {
+        source_table: 'projects',
+        target_table: 'members',
+        polymorphic: true,
+        polymorphic_type_column: 'source_type',
+        has_ar_forward: true,
+        forward_association: {
+          name: 'project_members',
+          type: 'has_many',
+          model: 'Project'
+        }
+      }
+
+      edges = [member_belongs_to_edge, namespace_has_many_edge, project_has_many_edge]
+
+      # Simulate the polymorphic_json task logic
+      polymorphic_data = {}
+
+      edges.each do |edge|
+        next unless edge[:polymorphic] && edge[:polymorphic_type_column]
+
+        # For has_many/has_one side (as: :something)
+        if edge[:has_ar_forward] && edge[:forward_association]
+          table_name = edge[:source_table]
+          type_column = edge[:polymorphic_type_column]
+
+          polymorphic_data[table_name] ||= {}
+          polymorphic_data[table_name][type_column] ||= Set.new
+        end
+
+        # For belongs_to side (polymorphic: true)
+        next unless edge[:has_ar_reverse] && edge[:reverse_association]
+
+        type_column = edge[:polymorphic_type_column]
+        source_model = edge[:reverse_association][:model]
+
+        # Look for the corresponding has_many/has_one relationships
+        edges.each do |other_edge|
+          next unless other_edge[:polymorphic] &&
+            other_edge[:polymorphic_type_column] == type_column &&
+            other_edge[:has_ar_forward]
+
+          target_table = other_edge[:source_table]
+
+          polymorphic_data[target_table] ||= {}
+          polymorphic_data[target_table][type_column] ||= Set.new
+          polymorphic_data[target_table][type_column] << source_model
+        end
+      end
+
+      # Convert sets to sorted arrays
+      output = {}
+      polymorphic_data.each do |table_name, columns|
+        output[table_name] = {}
+        columns.each do |type_column, models_set|
+          output[table_name][type_column] = models_set.to_a.sort
+        end
+      end
+
+      # Verify that Member models appear in the output
+      expect(output).to have_key('namespaces')
+      expect(output['namespaces']).to have_key('source_type')
+      expect(output['namespaces']['source_type']).to include('GroupMember')
+
+      expect(output).to have_key('projects')
+      expect(output['projects']).to have_key('source_type')
+      expect(output['projects']['source_type']).to include('GroupMember')
+    end
+  end
+
+  describe 'Member model polymorphic association' do
+    it 'has the expected polymorphic belongs_to :source association' do
+      expect(Member.reflections).to have_key('source')
+
+      source_reflection = Member.reflections['source']
+      expect(source_reflection.macro).to eq(:belongs_to)
+      expect(source_reflection.polymorphic?).to be true
+      expect(source_reflection.foreign_type).to eq('source_type')
+      expect(source_reflection.foreign_key).to eq('source_id')
+    end
+  end
+
+  describe 'Group model polymorphic associations' do
+    it 'has has_many associations with as: :source pointing to Member subclasses' do
+      group_member_associations = Group.reflections.select do |name, reflection|
+        reflection.macro == :has_many &&
+          reflection.options[:as] == :source &&
+          name.include?('member')
+      end
+
+      expect(group_member_associations).not_to be_empty
+
+      # Check that at least one association points to GroupMember or Member
+      has_member_association = group_member_associations.any? do |name, reflection|
+        reflection.class_name&.include?('Member')
+      end
+
+      expect(has_member_association).to be true
+    end
+  end
+
+  describe 'Project model polymorphic associations' do
+    it 'has has_many associations with as: :source pointing to Member subclasses' do
+      project_member_associations = Project.reflections.select do |name, reflection|
+        reflection.macro == :has_many &&
+          reflection.options[:as] == :source &&
+          name.include?('member')
+      end
+
+      expect(project_member_associations).not_to be_empty
+
+      # Check that at least one association points to ProjectMember or Member
+      has_member_association = project_member_associations.any? do |name, reflection|
+        reflection.class_name&.include?('Member')
+      end
+
+      expect(has_member_association).to be true
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/active_record_spec.rb b/spec/lib/gitlab/reflections/relationships/active_record_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..d5564cc1c968bf0d4ec6ff11aa3b9e42ed96c896
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/active_record_spec.rb
@@ -0,0 +1,443 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::ActiveRecord, feature_category: :database do
+  # Mock models reflection instance
+  let(:mock_models_reflection) do
+    double('models_reflection', models: mock_models)
+  end
+
+  let(:mock_models) do
+    [MockUserModel, MockProjectModel, MockIssueModel]
+  end
+
+  # Mock ActiveRecord models with reflections
+  class MockUserModel
+    def self.name
+      'User'
+    end
+
+    def self.table_name
+      'users'
+    end
+
+    def self.reflections
+      {
+        'projects' => MockHasManyReflection.new(:has_many, 'projects', 'Project'),
+        'profile' => MockHasOneReflection.new(:has_one, 'profile', 'Profile'),
+        'avatar' => MockHasOneAttachedReflection.new(:has_one_attached, 'avatar'),
+        'uploads' => MockHasManyAttachedReflection.new(:has_many_attached, 'uploads')
+      }
+    end
+  end
+
+  class MockProjectModel
+    def self.name
+      'Project'
+    end
+
+    def self.table_name
+      'projects'
+    end
+
+    def self.reflections
+      {
+        'owner' => MockBelongsToReflection.new(:belongs_to, 'owner', 'User'),
+        'issues' => MockHasManyReflection.new(:has_many, 'issues', 'Issue'),
+        'tags' => MockHabtmReflection.new(:has_and_belongs_to_many, 'tags', 'Tag')
+      }
+    end
+  end
+
+  class MockIssueModel
+    def self.name
+      'Issue'
+    end
+
+    def self.table_name
+      'issues'
+    end
+
+    def self.reflections
+      {
+        'project' => MockBelongsToReflection.new(:belongs_to, 'project', 'Project'),
+        'author' => MockBelongsToReflection.new(:belongs_to, 'author', 'User'),
+        'assignees' => MockHasManyReflection.new(:has_many, 'assignees', 'User'),
+        'unsupported_relation' => MockUnsupportedReflection.new(:custom_macro, 'unsupported', 'Other')
+      }
+    end
+  end
+
+  # Mock reflection classes
+  class MockBelongsToReflection
+    attr_reader :macro, :name, :class_name
+
+    def initialize(macro, name, class_name)
+      @macro = macro
+      @name = name
+      @class_name = class_name
+    end
+
+    def polymorphic?
+      false
+    end
+
+    def foreign_key
+      "#{name}_id"
+    end
+
+    def options
+      {}
+    end
+  end
+
+  class MockHasManyReflection
+    attr_reader :macro, :name, :class_name
+
+    def initialize(macro, name, class_name)
+      @macro = macro
+      @name = name
+      @class_name = class_name
+    end
+
+    def polymorphic?
+      false
+    end
+
+    def foreign_key
+      "#{name.singularize}_id"
+    end
+
+    def options
+      {}
+    end
+  end
+
+  class MockHasOneReflection
+    attr_reader :macro, :name, :class_name
+
+    def initialize(macro, name, class_name)
+      @macro = macro
+      @name = name
+      @class_name = class_name
+    end
+
+    def polymorphic?
+      false
+    end
+
+    def foreign_key
+      "#{name}_id"
+    end
+
+    def options
+      {}
+    end
+  end
+
+  class MockHabtmReflection
+    attr_reader :macro, :name, :class_name
+
+    def initialize(macro, name, class_name)
+      @macro = macro
+      @name = name
+      @class_name = class_name
+    end
+
+    def options
+      { join_table: "#{name}_join_table" }
+    end
+  end
+
+  class MockHasOneAttachedReflection
+    attr_reader :macro, :name
+
+    def initialize(macro, name)
+      @macro = macro
+      @name = name
+    end
+
+    def options
+      {}
+    end
+  end
+
+  class MockHasManyAttachedReflection
+    attr_reader :macro, :name
+
+    def initialize(macro, name)
+      @macro = macro
+      @name = name
+    end
+
+    def options
+      {}
+    end
+  end
+
+  class MockUnsupportedReflection
+    attr_reader :macro, :name, :class_name
+
+    def initialize(macro, name, class_name)
+      @macro = macro
+      @name = name
+      @class_name = class_name
+    end
+  end
+
+  # Mock handler classes
+  let(:mock_belongs_to_handler) { double('belongs_to_handler') }
+  let(:mock_has_association_handler) { double('has_association_handler') }
+  let(:mock_habtm_handler) { double('habtm_handler') }
+  let(:mock_active_storage_handler) { double('active_storage_handler') }
+
+  before do
+    # Stub the models reflection
+    allow(Gitlab::Reflections::Models::ActiveRecord).to receive(:instance).and_return(mock_models_reflection)
+
+    # Stub handler classes
+    stub_const('Gitlab::Reflections::Relationships::Handlers::BelongsToHandler', double('BelongsToHandler'))
+    stub_const('Gitlab::Reflections::Relationships::Handlers::HasAssociationHandler', double('HasAssociationHandler'))
+    stub_const('Gitlab::Reflections::Relationships::Handlers::HabtmHandler', double('HabtmHandler'))
+    stub_const('Gitlab::Reflections::Relationships::Handlers::ActiveStorageHandler', double('ActiveStorageHandler'))
+
+    # Mock handler instantiation and behavior
+    allow(Gitlab::Reflections::Relationships::Handlers::BelongsToHandler).to receive(:new).and_return(mock_belongs_to_handler)
+    allow(Gitlab::Reflections::Relationships::Handlers::HasAssociationHandler).to receive(:new).and_return(mock_has_association_handler)
+    allow(Gitlab::Reflections::Relationships::Handlers::HabtmHandler).to receive(:new).and_return(mock_habtm_handler)
+    allow(Gitlab::Reflections::Relationships::Handlers::ActiveStorageHandler).to receive(:new).and_return(mock_active_storage_handler)
+
+    # Mock handler build_relationship methods
+    allow(mock_belongs_to_handler).to receive(:build_relationship).and_return({ type: 'belongs_to',
+source_table: 'issues', target_table: 'projects' })
+    allow(mock_has_association_handler).to receive(:build_relationship).and_return({ type: 'has_many',
+source_table: 'projects', target_table: 'issues' })
+    allow(mock_habtm_handler).to receive(:build_relationship).and_return({ type: 'has_and_belongs_to_many',
+source_table: 'projects', target_table: 'tags' })
+    allow(mock_active_storage_handler).to receive(:build_relationship).and_return({ type: 'active_storage',
+source_table: 'users', target_table: 'active_storage_attachments' })
+  end
+
+  describe '#initialize' do
+    it 'initializes with models reflection instance' do
+      instance = described_class.new
+      expect(Gitlab::Reflections::Models::ActiveRecord).to have_received(:instance)
+    end
+  end
+
+  describe '#extract_relationships' do
+    subject { described_class.new.extract_relationships }
+
+    it 'returns an array of relationships' do
+      expect(subject).to be_an(Array)
+    end
+
+    it 'processes all supported relationship types' do
+      relationships = subject
+
+      # Should process belongs_to relationships
+      expect(mock_belongs_to_handler).to have_received(:build_relationship).at_least(:once)
+
+      # Should process has_many/has_one relationships
+      expect(mock_has_association_handler).to have_received(:build_relationship).at_least(:once)
+
+      # Should process has_and_belongs_to_many relationships
+      expect(mock_habtm_handler).to have_received(:build_relationship).at_least(:once)
+
+      # Should process Active Storage relationships
+      expect(mock_active_storage_handler).to have_received(:build_relationship).at_least(:once)
+    end
+
+    it 'skips unsupported relationship types' do
+      # The unsupported_relation from MockIssueModel should be skipped
+      # We can verify this by checking that no handler was created for it
+      relationships = subject
+
+      # Verify that only supported relationships are processed
+      expect(relationships).to all(be_a(Hash))
+    end
+
+    it 'filters out nil relationships' do
+      # Mock one handler to return nil
+      allow(mock_belongs_to_handler).to receive(:build_relationship).and_return(nil)
+
+      relationships = subject
+      expect(relationships).not_to include(nil)
+    end
+  end
+
+  describe 'constants' do
+    describe 'SUPPORTED_RELATIONSHIPS' do
+      it 'includes all expected relationship types' do
+        expected_relationships = %w[
+          has_many has_one belongs_to has_and_belongs_to_many
+          has_many_attached has_one_attached
+        ]
+
+        expect(described_class::SUPPORTED_RELATIONSHIPS).to match_array(expected_relationships)
+      end
+
+      it 'contains only string values' do
+        described_class::SUPPORTED_RELATIONSHIPS.each do |relationship|
+          expect(relationship).to be_a(String)
+        end
+      end
+    end
+
+    describe 'RELATIONSHIP_HANDLERS' do
+      it 'maps relationship types to handler classes' do
+        expected_mappings = {
+          'belongs_to' => 'Gitlab::Reflections::Relationships::Handlers::BelongsToHandler',
+          'has_many' => 'Gitlab::Reflections::Relationships::Handlers::HasAssociationHandler',
+          'has_one' => 'Gitlab::Reflections::Relationships::Handlers::HasAssociationHandler',
+          'has_and_belongs_to_many' => 'Gitlab::Reflections::Relationships::Handlers::HabtmHandler',
+          'has_many_attached' => 'Gitlab::Reflections::Relationships::Handlers::ActiveStorageHandler',
+          'has_one_attached' => 'Gitlab::Reflections::Relationships::Handlers::ActiveStorageHandler'
+        }
+
+        expected_mappings.each do |relationship_type, expected_handler_name|
+          handler_class = described_class::RELATIONSHIP_HANDLERS[relationship_type]
+          expect(handler_class.to_s).to eq(expected_handler_name)
+        end
+      end
+
+      it 'has handlers for all supported relationships' do
+        described_class::SUPPORTED_RELATIONSHIPS.each do |relationship|
+          expect(described_class::RELATIONSHIP_HANDLERS).to have_key(relationship)
+        end
+      end
+    end
+  end
+
+  describe 'private methods' do
+    let(:instance) { described_class.new }
+
+    describe '#get_source_data' do
+      it 'returns models from the models reflection' do
+        source_data = instance.send(:get_source_data)
+        expect(source_data).to eq(mock_models)
+      end
+    end
+
+    describe '#filter_source_data' do
+      it 'returns models unchanged' do
+        filtered_data = instance.send(:filter_source_data, mock_models)
+        expect(filtered_data).to eq(mock_models)
+      end
+    end
+
+    describe '#map_to_relationships' do
+      it 'processes reflections from all models' do
+        relationships = instance.send(:map_to_relationships, mock_models)
+
+        expect(relationships).to be_an(Array)
+        expect(relationships).not_to be_empty
+      end
+
+      it 'creates relationships for supported reflection types' do
+        relationships = instance.send(:map_to_relationships, [MockUserModel])
+
+        # Should create relationships for has_many, has_one, and Active Storage attachments
+        expect(relationships.length).to be >= 4 # projects, profile, avatar, uploads
+      end
+
+      it 'skips unsupported reflection types' do
+        relationships = instance.send(:map_to_relationships, [MockIssueModel])
+
+        # Should not include the unsupported_relation
+        expect(relationships).to all(be_a(Hash))
+      end
+    end
+
+    describe '#build_relationship_for_reflection' do
+      it 'creates appropriate handler for belongs_to reflection' do
+        reflection = MockBelongsToReflection.new(:belongs_to, 'project', 'Project')
+
+        relationship = instance.send(:build_relationship_for_reflection, MockIssueModel, 'project', reflection)
+
+        expect(Gitlab::Reflections::Relationships::Handlers::BelongsToHandler).to have_received(:new)
+          .with(MockIssueModel, 'project', reflection)
+        expect(relationship).to eq({ type: 'belongs_to', source_table: 'issues', target_table: 'projects' })
+      end
+
+      it 'creates appropriate handler for has_many reflection' do
+        reflection = MockHasManyReflection.new(:has_many, 'issues', 'Issue')
+
+        relationship = instance.send(:build_relationship_for_reflection, MockProjectModel, 'issues', reflection)
+
+        expect(Gitlab::Reflections::Relationships::Handlers::HasAssociationHandler).to have_received(:new)
+          .with(MockProjectModel, 'issues', reflection)
+        expect(relationship).to eq({ type: 'has_many', source_table: 'projects', target_table: 'issues' })
+      end
+
+      it 'creates appropriate handler for has_one reflection' do
+        reflection = MockHasOneReflection.new(:has_one, 'profile', 'Profile')
+
+        relationship = instance.send(:build_relationship_for_reflection, MockUserModel, 'profile', reflection)
+
+        expect(Gitlab::Reflections::Relationships::Handlers::HasAssociationHandler).to have_received(:new)
+          .with(MockUserModel, 'profile', reflection)
+        expect(relationship).to eq({ type: 'has_many', source_table: 'projects', target_table: 'issues' })
+      end
+
+      it 'creates appropriate handler for has_and_belongs_to_many reflection' do
+        reflection = MockHabtmReflection.new(:has_and_belongs_to_many, 'tags', 'Tag')
+
+        relationship = instance.send(:build_relationship_for_reflection, MockProjectModel, 'tags', reflection)
+
+        expect(Gitlab::Reflections::Relationships::Handlers::HabtmHandler).to have_received(:new)
+          .with(MockProjectModel, 'tags', reflection)
+        expect(relationship).to eq({ type: 'has_and_belongs_to_many', source_table: 'projects', target_table: 'tags' })
+      end
+
+      it 'creates appropriate handler for Active Storage reflections' do
+        reflection = MockHasOneAttachedReflection.new(:has_one_attached, 'avatar')
+
+        relationship = instance.send(:build_relationship_for_reflection, MockUserModel, 'avatar', reflection)
+
+        expect(Gitlab::Reflections::Relationships::Handlers::ActiveStorageHandler).to have_received(:new)
+          .with(MockUserModel, 'avatar', reflection)
+        expect(relationship).to eq({ type: 'active_storage', source_table: 'users',
+target_table: 'active_storage_attachments' })
+      end
+
+      it 'returns nil for unsupported reflection types' do
+        reflection = MockUnsupportedReflection.new(:custom_macro, 'unsupported', 'Other')
+
+        relationship = instance.send(:build_relationship_for_reflection, MockIssueModel, 'unsupported', reflection)
+
+        expect(relationship).to be_nil
+      end
+
+      it 'returns nil when handler returns nil' do
+        reflection = MockBelongsToReflection.new(:belongs_to, 'project', 'Project')
+        allow(mock_belongs_to_handler).to receive(:build_relationship).and_return(nil)
+
+        relationship = instance.send(:build_relationship_for_reflection, MockIssueModel, 'project', reflection)
+
+        expect(relationship).to be_nil
+      end
+    end
+  end
+
+  describe 'integration with handlers' do
+    it 'passes correct parameters to handlers' do
+      instance = described_class.new
+      instance.extract_relationships
+
+      # Verify that handlers are called with correct parameters
+      expect(Gitlab::Reflections::Relationships::Handlers::BelongsToHandler).to have_received(:new)
+        .with(MockProjectModel, 'owner', instance_of(MockBelongsToReflection))
+
+      expect(Gitlab::Reflections::Relationships::Handlers::HasAssociationHandler).to have_received(:new)
+        .with(MockUserModel, 'projects', instance_of(MockHasManyReflection))
+
+      expect(Gitlab::Reflections::Relationships::Handlers::HabtmHandler).to have_received(:new)
+        .with(MockProjectModel, 'tags', instance_of(MockHabtmReflection))
+
+      expect(Gitlab::Reflections::Relationships::Handlers::ActiveStorageHandler).to have_received(:new)
+        .with(MockUserModel, 'avatar', instance_of(MockHasOneAttachedReflection))
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/base_spec.rb b/spec/lib/gitlab/reflections/relationships/base_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..0d02ca9a918ae7d8698f140f98f1ccc0c5389621
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/base_spec.rb
@@ -0,0 +1,83 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Base do
+  let(:test_class) do
+    Class.new(described_class) do
+      private
+
+      def get_source_data
+        [{ name: 'test_data' }]
+      end
+
+      def filter_source_data(data)
+        data
+      end
+
+      def map_to_relationships(data)
+        data.map do |item|
+          {
+            source_table: 'test_table',
+            target_table: 'target_table',
+            foreign_key: 'test_id',
+            relationship_type: 'one_to_many'
+          }
+        end
+      end
+    end
+  end
+
+  describe '#extract_relationships' do
+    it 'returns relationship data from subclass implementation' do
+      extractor = test_class.new
+      relationships = extractor.extract_relationships
+      
+      expect(relationships).to have(1).item
+      expect(relationships.first[:source_table]).to eq('test_table')
+      expect(relationships.first[:target_table]).to eq('target_table')
+      expect(relationships.first[:foreign_key]).to eq('test_id')
+    end
+  end
+
+  describe '#build_relationship' do
+    let(:extractor) { test_class.new }
+
+    it 'returns a clean relationship hash for valid data' do
+      relationship = extractor.send(:build_relationship,
+        source_table: 'users',
+        target_table: 'posts',
+        foreign_key: 'user_id',
+        extra_field: 'value',
+        nil_field: nil
+      )
+      
+      expect(relationship).to eq({
+        source_table: 'users',
+        target_table: 'posts',
+        foreign_key: 'user_id',
+        extra_field: 'value'
+      })
+    end
+
+    it 'returns nil for missing required fields' do
+      relationship = extractor.send(:build_relationship,
+        source_table: 'users',
+        target_table: 'posts'
+        # missing foreign_key
+      )
+      
+      expect(relationship).to be_nil
+    end
+
+    it 'returns nil for empty required fields' do
+      relationship = extractor.send(:build_relationship,
+        source_table: 'users',
+        target_table: '',
+        foreign_key: 'user_id'
+      )
+      
+      expect(relationship).to be_nil
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/lib/gitlab/reflections/relationships/foreign_key_spec.rb b/spec/lib/gitlab/reflections/relationships/foreign_key_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..f046094a097e4e01d482e736e542543029560bcf
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/foreign_key_spec.rb
@@ -0,0 +1,392 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::ForeignKey, feature_category: :database do
+  # Mock PostgresForeignKey objects
+  let(:mock_pg_fk_1) do
+    double('pg_fk_1',
+      constrained_table_name: 'issues',
+      referenced_table_name: 'projects',
+      constrained_columns: ['project_id'],
+      referenced_columns: ['id'],
+      name: 'fk_issues_project_id',
+      on_delete_action: 'cascade',
+      on_update_action: 'restrict'
+    )
+  end
+
+  let(:mock_pg_fk_2) do
+    double('pg_fk_2',
+      constrained_table_name: 'merge_requests',
+      referenced_table_name: 'users',
+      constrained_columns: ['author_id'],
+      referenced_columns: ['id'],
+      name: 'fk_merge_requests_author_id',
+      on_delete_action: 'set_null',
+      on_update_action: 'cascade'
+    )
+  end
+
+  let(:mock_pg_fk_3) do
+    double('pg_fk_3',
+      constrained_table_name: 'notes',
+      referenced_table_name: 'projects',
+      constrained_columns: ['project_id'],
+      referenced_columns: ['id'],
+      name: 'fk_notes_project_id',
+      on_delete_action: 'restrict',
+      on_update_action: 'restrict'
+    )
+  end
+
+  # Mock scope that behaves like ActiveRecord relation
+  let(:mock_scope) do
+    double('scope').tap do |scope|
+      allow(scope).to receive(:find_each).and_yield(mock_pg_fk_1).and_yield(mock_pg_fk_2).and_yield(mock_pg_fk_3)
+    end
+  end
+
+  # Mock foreign key reflection
+  let(:mock_fk_reflection) do
+    double('fk_reflection', for_relationships: mock_scope)
+  end
+
+  describe '#initialize' do
+    context 'with custom fk_reflection' do
+      it 'uses the provided fk_reflection' do
+        instance = described_class.new(mock_fk_reflection)
+        expect(instance.instance_variable_get(:@fk_reflection)).to eq(mock_fk_reflection)
+      end
+    end
+
+    context 'without fk_reflection parameter' do
+      it 'uses default Gitlab::Reflections::Database::ForeignKeys' do
+        # Stub the default class
+        stub_const('Gitlab::Reflections::Database::ForeignKeys', mock_fk_reflection)
+
+        instance = described_class.new
+        expect(instance.instance_variable_get(:@fk_reflection)).to eq(mock_fk_reflection)
+      end
+    end
+  end
+
+  describe '#get_relationships' do
+    subject { described_class.new(mock_fk_reflection).get_relationships }
+
+    it 'returns an array of relationships' do
+      expect(subject).to be_an(Array)
+    end
+
+    it 'creates relationships for each foreign key' do
+      relationships = subject
+      expect(relationships.length).to eq(3)
+    end
+
+    it 'creates relationships with correct structure' do
+      relationships = subject
+
+      relationships.each do |relationship|
+        expect(relationship).to be_a(Hash)
+        expect(relationship).to have_key(:source_table)
+        expect(relationship).to have_key(:target_table)
+        expect(relationship).to have_key(:relationship_type)
+        expect(relationship).to have_key(:foreign_key)
+        expect(relationship).to have_key(:has_fk_constraint)
+        expect(relationship).to have_key(:fk_metadata)
+        expect(relationship).to have_key(:data_sources)
+      end
+    end
+
+    it 'sets relationship_type to one_to_many' do
+      relationships = subject
+
+      relationships.each do |relationship|
+        expect(relationship[:relationship_type]).to eq('one_to_many')
+      end
+    end
+
+    it 'sets has_fk_constraint to true' do
+      relationships = subject
+
+      relationships.each do |relationship|
+        expect(relationship[:has_fk_constraint]).to be true
+      end
+    end
+
+    it 'includes data_sources as foreign_key' do
+      relationships = subject
+
+      relationships.each do |relationship|
+        expect(relationship[:data_sources]).to eq(['foreign_key'])
+      end
+    end
+
+    it 'creates correct relationship for issues -> projects' do
+      relationships = subject
+      issues_project_rel = relationships.find { |r| r[:target_table] == 'issues' && r[:source_table] == 'projects' }
+
+      expect(issues_project_rel).not_to be_nil
+      expect(issues_project_rel[:foreign_key]).to eq('project_id')
+      expect(issues_project_rel[:foreign_key_table]).to eq('issues')
+      expect(issues_project_rel[:source_columns]).to eq(['id'])
+      expect(issues_project_rel[:target_columns]).to eq(['project_id'])
+      expect(issues_project_rel[:fk_metadata][:constraint_name]).to eq('fk_issues_project_id')
+      expect(issues_project_rel[:fk_metadata][:on_delete]).to eq('cascade')
+      expect(issues_project_rel[:fk_metadata][:on_update]).to eq('restrict')
+    end
+
+    it 'creates correct relationship for merge_requests -> users' do
+      relationships = subject
+      mr_user_rel = relationships.find { |r| r[:target_table] == 'merge_requests' && r[:source_table] == 'users' }
+
+      expect(mr_user_rel).not_to be_nil
+      expect(mr_user_rel[:foreign_key]).to eq('author_id')
+      expect(mr_user_rel[:foreign_key_table]).to eq('merge_requests')
+      expect(mr_user_rel[:source_columns]).to eq(['id'])
+      expect(mr_user_rel[:target_columns]).to eq(['author_id'])
+      expect(mr_user_rel[:fk_metadata][:constraint_name]).to eq('fk_merge_requests_author_id')
+      expect(mr_user_rel[:fk_metadata][:on_delete]).to eq('set_null')
+      expect(mr_user_rel[:fk_metadata][:on_update]).to eq('cascade')
+    end
+
+    it 'creates correct relationship for notes -> projects' do
+      relationships = subject
+      notes_project_rel = relationships.find { |r| r[:target_table] == 'notes' && r[:source_table] == 'projects' }
+
+      expect(notes_project_rel).not_to be_nil
+      expect(notes_project_rel[:foreign_key]).to eq('project_id')
+      expect(notes_project_rel[:foreign_key_table]).to eq('notes')
+      expect(notes_project_rel[:source_columns]).to eq(['id'])
+      expect(notes_project_rel[:target_columns]).to eq(['project_id'])
+      expect(notes_project_rel[:fk_metadata][:constraint_name]).to eq('fk_notes_project_id')
+      expect(notes_project_rel[:fk_metadata][:on_delete]).to eq('restrict')
+      expect(notes_project_rel[:fk_metadata][:on_update]).to eq('restrict')
+    end
+  end
+
+  describe 'private methods' do
+    let(:instance) { described_class.new(mock_fk_reflection) }
+
+    describe '#get_source_data' do
+      it 'converts PostgresForeignKey objects to internal format' do
+        source_data = instance.send(:get_source_data)
+
+        expect(source_data).to be_an(Array)
+        expect(source_data.length).to eq(3)
+
+        source_data.each do |item|
+          expect(item).to have_key(:table_name)
+          expect(item).to have_key(:foreign_key)
+          expect(item[:foreign_key]).to respond_to(:column)
+          expect(item[:foreign_key]).to respond_to(:to_table)
+          expect(item[:foreign_key]).to respond_to(:primary_key)
+          expect(item[:foreign_key]).to respond_to(:name)
+          expect(item[:foreign_key]).to respond_to(:on_delete)
+          expect(item[:foreign_key]).to respond_to(:on_update)
+        end
+      end
+
+      it 'calls find_each on the fk_reflection scope' do
+        expect(mock_scope).to receive(:find_each)
+        instance.send(:get_source_data)
+      end
+    end
+
+    describe '#filter_source_data' do
+      it 'returns data unchanged' do
+        test_data = [{ table_name: 'test', foreign_key: double }]
+        filtered_data = instance.send(:filter_source_data, test_data)
+
+        expect(filtered_data).to eq(test_data)
+      end
+    end
+
+    describe '#map_to_relationships' do
+      let(:test_items) do
+        [
+          {
+            table_name: 'issues',
+            foreign_key: OpenStruct.new(
+              column: 'project_id',
+              to_table: 'projects',
+              primary_key: 'id',
+              name: 'fk_test',
+              on_delete: 'cascade',
+              on_update: 'restrict'
+            )
+          }
+        ]
+      end
+
+      it 'maps items to relationships' do
+        relationships = instance.send(:map_to_relationships, test_items)
+
+        expect(relationships).to be_an(Array)
+        expect(relationships.length).to eq(1)
+        expect(relationships.first).to be_a(Hash)
+      end
+
+      it 'filters out nil relationships' do
+        # Mock build_foreign_key_relationship to return nil for one item
+        allow(instance).to receive(:build_foreign_key_relationship).and_return(nil, { test: 'relationship' })
+
+        test_items_with_nil = [
+          { table_name: 'table1', foreign_key: double },
+          { table_name: 'table2', foreign_key: double }
+        ]
+
+        relationships = instance.send(:map_to_relationships, test_items_with_nil)
+
+        expect(relationships).to eq([{ test: 'relationship' }])
+      end
+    end
+
+    describe '#build_foreign_key_relationship' do
+      let(:foreign_key) do
+        OpenStruct.new(
+          column: 'project_id',
+          to_table: 'projects',
+          primary_key: 'id',
+          name: 'fk_issues_project_id',
+          on_delete: 'cascade',
+          on_update: 'restrict'
+        )
+      end
+
+      it 'builds a complete relationship hash' do
+        relationship = instance.send(:build_foreign_key_relationship, 'issues', foreign_key)
+
+        expect(relationship).to be_a(Hash)
+        expect(relationship[:source_table]).to eq('projects')
+        expect(relationship[:target_table]).to eq('issues')
+        expect(relationship[:relationship_type]).to eq('one_to_many')
+        expect(relationship[:foreign_key]).to eq('project_id')
+        expect(relationship[:foreign_key_table]).to eq('issues')
+        expect(relationship[:source_columns]).to eq(['id'])
+        expect(relationship[:target_columns]).to eq(['project_id'])
+        expect(relationship[:has_fk_constraint]).to be true
+        expect(relationship[:data_sources]).to eq(['foreign_key'])
+      end
+
+      it 'includes fk_metadata with constraint details' do
+        relationship = instance.send(:build_foreign_key_relationship, 'issues', foreign_key)
+
+        expect(relationship[:fk_metadata]).to be_a(Hash)
+        expect(relationship[:fk_metadata][:constraint_name]).to eq('fk_issues_project_id')
+        expect(relationship[:fk_metadata][:on_delete]).to eq('cascade')
+        expect(relationship[:fk_metadata][:on_update]).to eq('restrict')
+      end
+
+      it 'handles foreign key without explicit primary_key' do
+        fk_without_pk = OpenStruct.new(
+          column: 'user_id',
+          to_table: 'users',
+          primary_key: nil,
+          name: 'fk_test',
+          on_delete: 'restrict',
+          on_update: 'restrict'
+        )
+
+        relationship = instance.send(:build_foreign_key_relationship, 'issues', fk_without_pk)
+
+        expect(relationship[:source_columns]).to eq(['id'])
+      end
+    end
+
+    describe '#build_fk_definition_from_pg_fk' do
+      it 'converts PostgresForeignKey to OpenStruct with correct attributes' do
+        fk_definition = instance.send(:build_fk_definition_from_pg_fk, mock_pg_fk_1)
+
+        expect(fk_definition).to be_an(OpenStruct)
+        expect(fk_definition.column).to eq('project_id')
+        expect(fk_definition.to_table).to eq('projects')
+        expect(fk_definition.primary_key).to eq('id')
+        expect(fk_definition.name).to eq('fk_issues_project_id')
+        expect(fk_definition.on_delete).to eq('cascade')
+        expect(fk_definition.on_update).to eq('restrict')
+      end
+
+      it 'handles single column foreign keys' do
+        # Test assumes single column FK (takes first element)
+        fk_definition = instance.send(:build_fk_definition_from_pg_fk, mock_pg_fk_2)
+
+        expect(fk_definition.column).to eq('author_id')
+        expect(fk_definition.to_table).to eq('users')
+        expect(fk_definition.primary_key).to eq('id')
+      end
+
+      it 'preserves constraint metadata' do
+        fk_definition = instance.send(:build_fk_definition_from_pg_fk, mock_pg_fk_2)
+
+        expect(fk_definition.name).to eq('fk_merge_requests_author_id')
+        expect(fk_definition.on_delete).to eq('set_null')
+        expect(fk_definition.on_update).to eq('cascade')
+      end
+    end
+  end
+
+  describe 'integration with foreign key reflection' do
+    it 'calls for_relationships on the fk_reflection' do
+      expect(mock_fk_reflection).to receive(:for_relationships).and_return(mock_scope)
+
+      described_class.new(mock_fk_reflection).get_relationships
+    end
+
+    it 'processes all foreign keys from the scope' do
+      expect(mock_scope).to receive(:find_each).and_yield(mock_pg_fk_1).and_yield(mock_pg_fk_2).and_yield(mock_pg_fk_3)
+
+      relationships = described_class.new(mock_fk_reflection).get_relationships
+      expect(relationships.length).to eq(3)
+    end
+  end
+
+  describe 'error handling' do
+    context 'when foreign key has missing data' do
+      let(:incomplete_pg_fk) do
+        double('incomplete_pg_fk',
+          constrained_table_name: 'issues',
+          referenced_table_name: nil,
+          constrained_columns: ['project_id'],
+          referenced_columns: ['id'],
+          name: 'fk_incomplete',
+          on_delete_action: 'cascade',
+          on_update_action: 'restrict'
+        )
+      end
+
+      let(:incomplete_scope) do
+        double('incomplete_scope').tap do |scope|
+          allow(scope).to receive(:find_each).and_yield(incomplete_pg_fk)
+        end
+      end
+
+      let(:incomplete_fk_reflection) do
+        double('incomplete_fk_reflection', for_relationships: incomplete_scope)
+      end
+
+      it 'handles foreign keys with missing referenced table' do
+        expect do
+          described_class.new(incomplete_fk_reflection).get_relationships
+        end.not_to raise_error
+      end
+    end
+
+    context 'when scope is empty' do
+      let(:empty_scope) do
+        double('empty_scope').tap do |scope|
+          allow(scope).to receive(:find_each)
+        end
+      end
+
+      let(:empty_fk_reflection) do
+        double('empty_fk_reflection', for_relationships: empty_scope)
+      end
+
+      it 'returns empty array when no foreign keys exist' do
+        relationships = described_class.new(empty_fk_reflection).get_relationships
+        expect(relationships).to eq([])
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/handlers/active_storage_handler_spec.rb b/spec/lib/gitlab/reflections/relationships/handlers/active_storage_handler_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..250409e69ace8bcfa93ea521b87451e11e2fea92
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/handlers/active_storage_handler_spec.rb
@@ -0,0 +1,447 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Handlers::ActiveStorageHandler do
+  # Mock ActiveRecord model class with ActiveStorage attachments
+  let(:mock_model_class) do
+    Class.new do
+      def self.table_name
+        'users'
+      end
+
+      def self.name
+        'User'
+      end
+    end
+  end
+
+  # Mock ActiveStorage has_one_attached reflection
+  let(:mock_has_one_attached_reflection) do
+    double('has_one_attached_reflection',
+      macro: :has_one_attached,
+      name: :avatar,
+      klass: mock_attachment_class,
+      table_name: 'active_storage_attachments',
+      plural_name: 'active_storage_attachments',
+      polymorphic?: false,
+      foreign_type: nil,
+      options: {}
+    )
+  end
+
+  # Mock ActiveStorage has_many_attached reflection
+  let(:mock_has_many_attached_reflection) do
+    double('has_many_attached_reflection',
+      macro: :has_many_attached,
+      name: :documents,
+      klass: mock_attachment_class,
+      table_name: 'active_storage_attachments',
+      plural_name: 'active_storage_attachments',
+      polymorphic?: false,
+      foreign_type: nil,
+      options: {}
+    )
+  end
+
+  # Mock ActiveStorage::Attachment class
+  let(:mock_attachment_class) do
+    Class.new do
+      def self.table_name
+        'active_storage_attachments'
+      end
+
+      def self.name
+        'ActiveStorage::Attachment'
+      end
+    end
+  end
+
+  # Mock reflection that raises NameError for target table determination
+  let(:mock_problematic_reflection) do
+    double('problematic_reflection',
+      macro: :has_one_attached,
+      name: :unknown_attachment,
+      table_name: 'active_storage_attachments',
+      plural_name: 'active_storage_attachments',
+      polymorphic?: false,
+      foreign_type: nil,
+      options: {}
+    ).tap do |reflection|
+      allow(reflection).to receive(:klass).and_raise(NameError, 'uninitialized constant ActiveStorage::Attachment')
+    end
+  end
+
+  let(:association_name) { :avatar }
+  let(:handler) { described_class.new(mock_model_class, association_name, mock_has_one_attached_reflection) }
+
+  describe '#initialize' do
+    it 'inherits from BaseHandler' do
+      expect(described_class.superclass).to eq(Gitlab::Reflections::Relationships::Handlers::BaseHandler)
+    end
+
+    it 'initializes with model, association name, and reflection' do
+      expect(handler.instance_variable_get(:@model)).to eq(mock_model_class)
+      expect(handler.instance_variable_get(:@association_name)).to eq(association_name)
+      expect(handler.instance_variable_get(:@reflection)).to eq(mock_has_one_attached_reflection)
+    end
+  end
+
+  describe '#build_relationship' do
+    context 'with has_one_attached reflection' do
+      it 'builds a complete ActiveStorage relationship hash' do
+        result = handler.build_relationship
+
+        expect(result).to be_a(Hash)
+        expect(result[:source_table]).to eq('users')
+        expect(result[:target_table]).to eq('active_storage_attachments')
+        expect(result[:relationship_type]).to eq('one_to_one')
+        expect(result[:foreign_key]).to eq('record_id')
+        expect(result[:has_ar_forward]).to be true
+        expect(result[:polymorphic]).to be false
+        expect(result[:data_sources]).to eq(['active_record'])
+      end
+
+      it 'includes forward association details' do
+        result = handler.build_relationship
+
+        expect(result[:forward_association]).to be_a(Hash)
+        expect(result[:forward_association][:name]).to eq('avatar')
+        expect(result[:forward_association][:type]).to eq('has_one_attached')
+        expect(result[:forward_association][:model]).to eq('User')
+      end
+
+      it 'sets the correct relationship type for has_one_attached' do
+        result = handler.build_relationship
+        expect(result[:relationship_type]).to eq('one_to_one')
+      end
+    end
+
+    context 'with has_many_attached reflection' do
+      let(:many_handler) { described_class.new(mock_model_class, :documents, mock_has_many_attached_reflection) }
+
+      it 'builds a complete ActiveStorage relationship hash' do
+        result = many_handler.build_relationship
+
+        expect(result).to be_a(Hash)
+        expect(result[:source_table]).to eq('users')
+        expect(result[:target_table]).to eq('active_storage_attachments')
+        expect(result[:relationship_type]).to eq('one_to_many')
+        expect(result[:foreign_key]).to eq('record_id')
+        expect(result[:has_ar_forward]).to be true
+        expect(result[:polymorphic]).to be false
+        expect(result[:data_sources]).to eq(['active_record'])
+      end
+
+      it 'includes forward association details' do
+        result = many_handler.build_relationship
+
+        expect(result[:forward_association]).to be_a(Hash)
+        expect(result[:forward_association][:name]).to eq('documents')
+        expect(result[:forward_association][:type]).to eq('has_many_attached')
+        expect(result[:forward_association][:model]).to eq('User')
+      end
+
+      it 'sets the correct relationship type for has_many_attached' do
+        result = many_handler.build_relationship
+        expect(result[:relationship_type]).to eq('one_to_many')
+      end
+    end
+
+    context 'when target table determination fails' do
+      let(:problematic_handler) { described_class.new(mock_model_class, :unknown_attachment, mock_problematic_reflection) }
+
+      it 'falls back to reflection table_name or plural_name' do
+        result = problematic_handler.build_relationship
+
+        expect(result[:target_table]).to eq('active_storage_attachments')
+        expect(result[:relationship_type]).to eq('one_to_one')
+      end
+    end
+
+    context 'when required fields are missing' do
+      let(:incomplete_reflection) do
+        double('incomplete_reflection',
+          macro: :has_one_attached,
+          name: :avatar,
+          klass: mock_attachment_class,
+          table_name: 'active_storage_attachments',
+          plural_name: 'active_storage_attachments',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:incomplete_handler) { described_class.new(mock_model_class, association_name, incomplete_reflection) }
+
+      before do
+        # Mock a scenario where source table is nil
+        allow(mock_model_class).to receive(:table_name).and_return(nil)
+        allow(Gitlab::AppLogger).to receive(:debug)
+      end
+
+      it 'returns nil when source_table is missing' do
+        result = incomplete_handler.build_relationship
+
+        expect(result).to be_nil
+        expect(Gitlab::AppLogger).to have_received(:debug).with(/Failed to build relationship: missing required fields/)
+      end
+    end
+
+    context 'when an exception occurs during relationship building' do
+      before do
+        allow(mock_model_class).to receive(:table_name).and_raise(StandardError, 'Database connection error')
+        allow(Gitlab::AppLogger).to receive(:debug)
+      end
+
+      it 'logs the error and returns nil' do
+        result = handler.build_relationship
+
+        expect(result).to be_nil
+        expect(Gitlab::AppLogger).to have_received(:debug).with('Failed to build relationship: Database connection error')
+      end
+    end
+  end
+
+  describe 'private methods' do
+    describe '#determine_target_table' do
+      it 'returns active_storage_attachments table' do
+        # Access private method for testing
+        target_table = handler.send(:determine_target_table)
+        expect(target_table).to eq('active_storage_attachments')
+      end
+
+      context 'when klass raises NameError' do
+        let(:problematic_handler) { described_class.new(mock_model_class, :unknown_attachment, mock_problematic_reflection) }
+
+        it 'falls back to reflection table_name' do
+          target_table = problematic_handler.send(:determine_target_table)
+          expect(target_table).to eq('active_storage_attachments')
+        end
+      end
+    end
+
+    describe '#base_attributes' do
+      context 'with has_one_attached' do
+        it 'sets relationship_type to one_to_one' do
+          base_attrs = handler.send(:base_attributes)
+
+          expect(base_attrs[:relationship_type]).to eq('one_to_one')
+          expect(base_attrs[:source_table]).to eq('users')
+          expect(base_attrs[:target_table]).to eq('active_storage_attachments')
+          expect(base_attrs[:has_ar_forward]).to be true
+          expect(base_attrs[:polymorphic]).to be false
+          expect(base_attrs[:data_sources]).to eq(['active_record'])
+        end
+      end
+
+      context 'with has_many_attached' do
+        let(:many_handler) { described_class.new(mock_model_class, :documents, mock_has_many_attached_reflection) }
+
+        it 'sets relationship_type to one_to_many' do
+          base_attrs = many_handler.send(:base_attributes)
+
+          expect(base_attrs[:relationship_type]).to eq('one_to_many')
+          expect(base_attrs[:source_table]).to eq('users')
+          expect(base_attrs[:target_table]).to eq('active_storage_attachments')
+          expect(base_attrs[:has_ar_forward]).to be true
+          expect(base_attrs[:polymorphic]).to be false
+          expect(base_attrs[:data_sources]).to eq(['active_record'])
+        end
+      end
+
+      it 'includes forward association information' do
+        base_attrs = handler.send(:base_attributes)
+
+        expect(base_attrs[:forward_association]).to be_a(Hash)
+        expect(base_attrs[:forward_association][:name]).to eq('avatar')
+        expect(base_attrs[:forward_association][:type]).to eq('has_one_attached')
+        expect(base_attrs[:forward_association][:model]).to eq('User')
+      end
+    end
+  end
+
+  # Integration tests with different ActiveStorage scenarios
+  describe 'integration scenarios' do
+    context 'with project logo attachment' do
+      let(:mock_project_model) do
+        Class.new do
+          def self.table_name
+            'projects'
+          end
+
+          def self.name
+            'Project'
+          end
+        end
+      end
+
+      let(:logo_reflection) do
+        double('logo_reflection',
+          macro: :has_one_attached,
+          name: :logo,
+          klass: mock_attachment_class,
+          table_name: 'active_storage_attachments',
+          plural_name: 'active_storage_attachments',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:logo_handler) { described_class.new(mock_project_model, :logo, logo_reflection) }
+
+      it 'correctly handles project logo attachment' do
+        result = logo_handler.build_relationship
+
+        expect(result[:source_table]).to eq('projects')
+        expect(result[:target_table]).to eq('active_storage_attachments')
+        expect(result[:foreign_key]).to eq('record_id')
+        expect(result[:relationship_type]).to eq('one_to_one')
+        expect(result[:forward_association][:name]).to eq('logo')
+      end
+    end
+
+    context 'with issue attachments' do
+      let(:mock_issue_model) do
+        Class.new do
+          def self.table_name
+            'issues'
+          end
+
+          def self.name
+            'Issue'
+          end
+        end
+      end
+
+      let(:attachments_reflection) do
+        double('attachments_reflection',
+          macro: :has_many_attached,
+          name: :attachments,
+          klass: mock_attachment_class,
+          table_name: 'active_storage_attachments',
+          plural_name: 'active_storage_attachments',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:attachments_handler) { described_class.new(mock_issue_model, :attachments, attachments_reflection) }
+
+      it 'correctly handles issue attachments' do
+        result = attachments_handler.build_relationship
+
+        expect(result[:source_table]).to eq('issues')
+        expect(result[:target_table]).to eq('active_storage_attachments')
+        expect(result[:foreign_key]).to eq('record_id')
+        expect(result[:relationship_type]).to eq('one_to_many')
+        expect(result[:forward_association][:name]).to eq('attachments')
+      end
+    end
+
+    context 'with user profile images' do
+      let(:profile_images_reflection) do
+        double('profile_images_reflection',
+          macro: :has_many_attached,
+          name: :profile_images,
+          klass: mock_attachment_class,
+          table_name: 'active_storage_attachments',
+          plural_name: 'active_storage_attachments',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:profile_images_handler) { described_class.new(mock_model_class, :profile_images, profile_images_reflection) }
+
+      it 'correctly handles user profile images' do
+        result = profile_images_handler.build_relationship
+
+        expect(result[:source_table]).to eq('users')
+        expect(result[:target_table]).to eq('active_storage_attachments')
+        expect(result[:foreign_key]).to eq('record_id')
+        expect(result[:relationship_type]).to eq('one_to_many')
+        expect(result[:forward_association][:name]).to eq('profile_images')
+      end
+    end
+  end
+
+  # Edge cases and error handling
+  describe 'edge cases' do
+    context 'with unknown macro type' do
+      let(:unknown_macro_reflection) do
+        double('unknown_macro_reflection',
+          macro: :unknown_attachment_type,
+          name: :unknown,
+          klass: mock_attachment_class,
+          table_name: 'active_storage_attachments',
+          plural_name: 'active_storage_attachments',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:unknown_handler) { described_class.new(mock_model_class, :unknown, unknown_macro_reflection) }
+
+      it 'defaults to unknown relationship type from base class' do
+        result = unknown_handler.build_relationship
+
+        # The base class determine_relationship_type method should handle unknown macros
+        expect(result[:relationship_type]).to eq('unknown')
+      end
+    end
+
+    context 'with nil attachment name' do
+      let(:nil_name_reflection) do
+        double('nil_name_reflection',
+          macro: :has_one_attached,
+          name: nil,
+          klass: mock_attachment_class,
+          table_name: 'active_storage_attachments',
+          plural_name: 'active_storage_attachments',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:nil_name_handler) { described_class.new(mock_model_class, nil, nil_name_reflection) }
+
+      it 'handles nil attachment name gracefully' do
+        result = nil_name_handler.build_relationship
+
+        expect(result).to be_a(Hash)
+        expect(result[:forward_association][:name]).to eq('')
+      end
+    end
+
+    context 'with empty table name' do
+      let(:empty_table_model) do
+        Class.new do
+          def self.table_name
+            ''
+          end
+
+          def self.name
+            'EmptyTableModel'
+          end
+        end
+      end
+
+      let(:empty_table_handler) { described_class.new(empty_table_model, :avatar, mock_has_one_attached_reflection) }
+
+      it 'handles empty table name gracefully' do
+        allow(Gitlab::AppLogger).to receive(:debug)
+        result = empty_table_handler.build_relationship
+
+        expect(result).to be_nil
+        expect(Gitlab::AppLogger).to have_received(:debug).with(/Failed to build relationship: missing required fields/)
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/lib/gitlab/reflections/relationships/handlers/base_handler_spec.rb b/spec/lib/gitlab/reflections/relationships/handlers/base_handler_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..d1e5e33a620db4565c05381c741c890841a90b5a
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/handlers/base_handler_spec.rb
@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Handlers::BaseHandler, feature_category: :database do
+  let(:model) { Note }
+  let(:association_name) { 'noteable' }
+  let(:reflection) { model.reflections[association_name.to_s] }
+  let(:handler) { described_class.new(model, association_name, reflection) }
+
+  describe '#build_relationship' do
+    it 'raises NotImplementedError' do
+      expect { handler.build_relationship }.to raise_error(NotImplementedError)
+    end
+  end
+
+  describe '#polymorphic?' do
+    it 'returns false for base handler' do
+      expect(handler.polymorphic?).to be false
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/handlers/belongs_to_handler_spec.rb b/spec/lib/gitlab/reflections/relationships/handlers/belongs_to_handler_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..d78f3d44f8555faacc5c1fc4ad53eb5ff9eb5022
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/handlers/belongs_to_handler_spec.rb
@@ -0,0 +1,39 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Handlers::BelongsToHandler, feature_category: :database do
+  let(:model) { Project }
+  let(:association_name) { 'namespace' }
+  let(:reflection) { model.reflections[association_name.to_s] }
+  let(:handler) { described_class.new(model, association_name, reflection) }
+
+  describe '#build_relationship' do
+    context 'with regular belongs_to association' do
+      it 'builds regular belongs_to relationship' do
+        relationship = handler.build_relationship
+
+        expect(relationship).to include(
+          source_table: 'namespaces',  # The referenced table (parent)
+          target_table: 'projects',    # The referencing table (child)
+          source_column: 'id',         # The referenced column
+          relationship_type: 'many_to_one',
+          polymorphic: false
+        )
+        expect(relationship[:reverse_association]).to include(
+          name: 'namespace',
+          type: 'belongs_to',
+          model: 'Project'
+        )
+        expect(relationship[:has_ar_reverse]).to be true
+      end
+
+      it 'returns same relationship for build_relationship_with_target' do
+        original_relationship = handler.build_relationship
+        target_relationship = handler.build_relationship_with_target('some_table')
+
+        expect(target_relationship).to eq(original_relationship)
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/handlers/habtm_handler_spec.rb b/spec/lib/gitlab/reflections/relationships/handlers/habtm_handler_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..8a00b445fdc412901a1f51e91942732dc34e5571
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/handlers/habtm_handler_spec.rb
@@ -0,0 +1,396 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Handlers::HabtmHandler do
+  # Mock ActiveRecord model class
+  let(:mock_model_class) do
+    Class.new do
+      def self.table_name
+        'users'
+      end
+
+      def self.name
+        'User'
+      end
+    end
+  end
+
+  # Mock HABTM reflection
+  let(:mock_habtm_reflection) do
+    double('habtm_reflection',
+      macro: :has_and_belongs_to_many,
+      foreign_key: 'user_id',
+      association_foreign_key: 'project_id',
+      join_table: 'projects_users',
+      klass: mock_target_model_class,
+      table_name: 'projects',
+      plural_name: 'projects',
+      polymorphic?: false,
+      foreign_type: nil,
+      options: {}
+    )
+  end
+
+  # Mock target model class
+  let(:mock_target_model_class) do
+    Class.new do
+      def self.table_name
+        'projects'
+      end
+
+      def self.name
+        'Project'
+      end
+    end
+  end
+
+  # Mock HABTM reflection with custom join table
+  let(:mock_custom_habtm_reflection) do
+    double('custom_habtm_reflection',
+      macro: :has_and_belongs_to_many,
+      foreign_key: 'user_id',
+      association_foreign_key: 'role_id',
+      join_table: 'user_roles',
+      klass: mock_role_model_class,
+      table_name: 'roles',
+      plural_name: 'roles',
+      polymorphic?: false,
+      foreign_type: nil,
+      options: {}
+    )
+  end
+
+  let(:mock_role_model_class) do
+    Class.new do
+      def self.table_name
+        'roles'
+      end
+
+      def self.name
+        'Role'
+      end
+    end
+  end
+
+  # Mock reflection that raises NameError for target table determination
+  let(:mock_problematic_reflection) do
+    double('problematic_reflection',
+      macro: :has_and_belongs_to_many,
+      foreign_key: 'user_id',
+      association_foreign_key: 'unknown_id',
+      join_table: 'users_unknowns',
+      table_name: 'unknowns',
+      plural_name: 'unknowns',
+      polymorphic?: false,
+      foreign_type: nil,
+      options: {}
+    ).tap do |reflection|
+      allow(reflection).to receive(:klass).and_raise(NameError, 'uninitialized constant Unknown')
+    end
+  end
+
+  let(:association_name) { :projects }
+  let(:handler) { described_class.new(mock_model_class, association_name, mock_habtm_reflection) }
+
+  describe '#initialize' do
+    it 'inherits from BaseHandler' do
+      expect(described_class.superclass).to eq(Gitlab::Reflections::Relationships::Handlers::BaseHandler)
+    end
+
+    it 'initializes with model, association name, and reflection' do
+      expect(handler.instance_variable_get(:@model)).to eq(mock_model_class)
+      expect(handler.instance_variable_get(:@association_name)).to eq(association_name)
+      expect(handler.instance_variable_get(:@reflection)).to eq(mock_habtm_reflection)
+    end
+  end
+
+  describe '#build_relationship' do
+    context 'with valid HABTM reflection' do
+      it 'builds a complete HABTM relationship hash' do
+        result = handler.build_relationship
+
+        expect(result).to be_a(Hash)
+        expect(result[:source_table]).to eq('users')
+        expect(result[:target_table]).to eq('projects')
+        expect(result[:relationship_type]).to eq('many_to_many')
+        expect(result[:foreign_key]).to eq('user_id')
+        expect(result[:foreign_key_table]).to eq('projects_users')
+        expect(result[:through_table]).to eq('projects_users')
+        expect(result[:has_ar_forward]).to be true
+        expect(result[:polymorphic]).to be false
+        expect(result[:data_sources]).to eq(['active_record'])
+      end
+
+      it 'includes forward association details' do
+        result = handler.build_relationship
+
+        expect(result[:forward_association]).to be_a(Hash)
+        expect(result[:forward_association][:name]).to eq('projects')
+        expect(result[:forward_association][:type]).to eq('has_and_belongs_to_many')
+        expect(result[:forward_association][:model]).to eq('User')
+      end
+
+      it 'sets the correct relationship type for HABTM' do
+        result = handler.build_relationship
+        expect(result[:relationship_type]).to eq('many_to_many')
+      end
+    end
+
+    context 'with custom join table' do
+      let(:custom_handler) { described_class.new(mock_model_class, :roles, mock_custom_habtm_reflection) }
+
+      it 'uses the custom join table' do
+        result = custom_handler.build_relationship
+
+        expect(result[:foreign_key_table]).to eq('user_roles')
+        expect(result[:through_table]).to eq('user_roles')
+        expect(result[:target_table]).to eq('roles')
+      end
+    end
+
+    context 'when target table determination fails' do
+      let(:problematic_handler) { described_class.new(mock_model_class, :unknowns, mock_problematic_reflection) }
+
+      it 'falls back to reflection table_name or plural_name' do
+        result = problematic_handler.build_relationship
+
+        expect(result[:target_table]).to eq('unknowns')
+        expect(result[:relationship_type]).to eq('many_to_many')
+      end
+    end
+
+    context 'when required fields are missing' do
+      let(:incomplete_reflection) do
+        double('incomplete_reflection',
+          macro: :has_and_belongs_to_many,
+          foreign_key: nil,
+          association_foreign_key: 'project_id',
+          join_table: 'projects_users',
+          klass: mock_target_model_class,
+          table_name: 'projects',
+          plural_name: 'projects',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:incomplete_handler) { described_class.new(mock_model_class, association_name, incomplete_reflection) }
+
+      it 'returns nil when foreign_key is missing' do
+        allow(Gitlab::AppLogger).to receive(:debug)
+        result = incomplete_handler.build_relationship
+
+        expect(result).to be_nil
+        expect(Gitlab::AppLogger).to have_received(:debug).with(/Failed to build relationship: missing required fields/)
+      end
+    end
+
+    context 'when an exception occurs during relationship building' do
+      before do
+        allow(mock_habtm_reflection).to receive(:foreign_key).and_raise(StandardError, 'Database connection error')
+        allow(Gitlab::AppLogger).to receive(:debug)
+      end
+
+      it 'logs the error and returns nil' do
+        result = handler.build_relationship
+
+        expect(result).to be_nil
+        expect(Gitlab::AppLogger).to have_received(:debug).with('Failed to build relationship: Database connection error')
+      end
+    end
+  end
+
+  describe 'private methods' do
+    describe '#base_attributes' do
+      it 'overrides relationship_type to many_to_many' do
+        # Access private method for testing
+        base_attrs = handler.send(:base_attributes)
+
+        expect(base_attrs[:relationship_type]).to eq('many_to_many')
+        expect(base_attrs[:source_table]).to eq('users')
+        expect(base_attrs[:target_table]).to eq('projects')
+        expect(base_attrs[:has_ar_forward]).to be true
+        expect(base_attrs[:polymorphic]).to be false
+        expect(base_attrs[:data_sources]).to eq(['active_record'])
+      end
+
+      it 'includes forward association information' do
+        base_attrs = handler.send(:base_attributes)
+
+        expect(base_attrs[:forward_association]).to be_a(Hash)
+        expect(base_attrs[:forward_association][:name]).to eq('projects')
+        expect(base_attrs[:forward_association][:type]).to eq('has_and_belongs_to_many')
+        expect(base_attrs[:forward_association][:model]).to eq('User')
+      end
+    end
+  end
+
+  # Integration tests with different HABTM scenarios
+  describe 'integration scenarios' do
+    context 'with tags and taggables HABTM relationship' do
+      let(:mock_tag_model) do
+        Class.new do
+          def self.table_name
+            'tags'
+          end
+
+          def self.name
+            'Tag'
+          end
+        end
+      end
+
+      let(:mock_taggable_model) do
+        Class.new do
+          def self.table_name
+            'issues'
+          end
+
+          def self.name
+            'Issue'
+          end
+        end
+      end
+
+      let(:tags_reflection) do
+        double('tags_reflection',
+          macro: :has_and_belongs_to_many,
+          foreign_key: 'issue_id',
+          association_foreign_key: 'tag_id',
+          join_table: 'issue_tags',
+          klass: mock_tag_model,
+          table_name: 'tags',
+          plural_name: 'tags',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:tags_handler) { described_class.new(mock_taggable_model, :tags, tags_reflection) }
+
+      it 'correctly handles tags relationship' do
+        result = tags_handler.build_relationship
+
+        expect(result[:source_table]).to eq('issues')
+        expect(result[:target_table]).to eq('tags')
+        expect(result[:foreign_key]).to eq('issue_id')
+        expect(result[:through_table]).to eq('issue_tags')
+        expect(result[:relationship_type]).to eq('many_to_many')
+      end
+    end
+
+    context 'with permissions and roles HABTM relationship' do
+      let(:mock_permission_model) do
+        Class.new do
+          def self.table_name
+            'permissions'
+          end
+
+          def self.name
+            'Permission'
+          end
+        end
+      end
+
+      let(:mock_role_model) do
+        Class.new do
+          def self.table_name
+            'roles'
+          end
+
+          def self.name
+            'Role'
+          end
+        end
+      end
+
+      let(:permissions_reflection) do
+        double('permissions_reflection',
+          macro: :has_and_belongs_to_many,
+          foreign_key: 'role_id',
+          association_foreign_key: 'permission_id',
+          join_table: 'permissions_roles',
+          klass: mock_permission_model,
+          table_name: 'permissions',
+          plural_name: 'permissions',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:permissions_handler) { described_class.new(mock_role_model, :permissions, permissions_reflection) }
+
+      it 'correctly handles permissions relationship' do
+        result = permissions_handler.build_relationship
+
+        expect(result[:source_table]).to eq('roles')
+        expect(result[:target_table]).to eq('permissions')
+        expect(result[:foreign_key]).to eq('role_id')
+        expect(result[:through_table]).to eq('permissions_roles')
+        expect(result[:relationship_type]).to eq('many_to_many')
+      end
+    end
+  end
+
+  # Edge cases and error handling
+  describe 'edge cases' do
+    context 'with empty join table name' do
+      let(:empty_join_reflection) do
+        double('empty_join_reflection',
+          macro: :has_and_belongs_to_many,
+          foreign_key: 'user_id',
+          association_foreign_key: 'project_id',
+          join_table: '',
+          klass: mock_target_model_class,
+          table_name: 'projects',
+          plural_name: 'projects',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:empty_join_handler) { described_class.new(mock_model_class, association_name, empty_join_reflection) }
+
+      it 'handles empty join table gracefully' do
+        allow(Gitlab::AppLogger).to receive(:debug)
+        result = empty_join_handler.build_relationship
+
+        expect(result).to be_nil
+        expect(Gitlab::AppLogger).to have_received(:debug).with(/Failed to build relationship: missing required fields/)
+      end
+    end
+
+    context 'with nil association foreign key' do
+      let(:nil_assoc_fk_reflection) do
+        double('nil_assoc_fk_reflection',
+          macro: :has_and_belongs_to_many,
+          foreign_key: 'user_id',
+          association_foreign_key: nil,
+          join_table: 'projects_users',
+          klass: mock_target_model_class,
+          table_name: 'projects',
+          plural_name: 'projects',
+          polymorphic?: false,
+          foreign_type: nil,
+          options: {}
+        )
+      end
+
+      let(:nil_assoc_fk_handler) { described_class.new(mock_model_class, association_name, nil_assoc_fk_reflection) }
+
+      it 'still builds relationship without association foreign key in main hash' do
+        result = nil_assoc_fk_handler.build_relationship
+
+        expect(result).to be_a(Hash)
+        expect(result[:source_table]).to eq('users')
+        expect(result[:target_table]).to eq('projects')
+        expect(result[:relationship_type]).to eq('many_to_many')
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/lib/gitlab/reflections/relationships/handlers/has_association_handler_spec.rb b/spec/lib/gitlab/reflections/relationships/handlers/has_association_handler_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..1ef95f36bbd2f5875d0696340b4d03688b9165d7
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/handlers/has_association_handler_spec.rb
@@ -0,0 +1,62 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Handlers::HasAssociationHandler, feature_category: :database do
+  let(:model) { Project }
+  let(:association_name) { 'issues' }
+  let(:reflection) { model.reflections[association_name.to_s] }
+  let(:handler) { described_class.new(model, association_name, reflection) }
+
+  describe '#build_relationship' do
+    context 'with regular has_many association' do
+      it 'builds regular has_many relationship' do
+        relationship = handler.build_relationship
+
+        expect(relationship).to include(
+          source_table: 'projects',
+          target_table: 'issues',
+          relationship_type: 'one_to_many',
+          polymorphic: false
+        )
+        expect(relationship[:forward_association]).to include(
+          name: 'issues',
+          type: 'has_many',
+          model: 'Project'
+        )
+        expect(relationship[:has_ar_forward]).to be true
+      end
+
+      it 'returns same relationship for build_relationship_with_target' do
+        original_relationship = handler.build_relationship
+        target_relationship = handler.build_relationship_with_target('some_table')
+
+        expect(target_relationship).to eq(original_relationship)
+      end
+    end
+
+    context 'with has_many through association' do
+      let(:model) { User }
+      let(:association_name) { 'assigned_issues' }
+      let(:reflection) { model.reflections[association_name.to_s] }
+
+      it 'builds has_many through relationship' do
+        relationship = handler.build_relationship
+
+        expect(relationship).to include(
+          source_table: 'users',
+          target_table: 'issues',
+          relationship_type: 'one_to_many',
+          polymorphic: false
+        )
+        expect(relationship[:through_table]).to eq('issue_assignees')
+        expect(relationship[:forward_association]).to include(
+          name: 'assigned_issues',
+          type: 'has_many',
+          model: 'User'
+        )
+        expect(relationship[:has_ar_forward]).to be true
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/handlers/polymorphic_belongs_to_handler_spec.rb b/spec/lib/gitlab/reflections/relationships/handlers/polymorphic_belongs_to_handler_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..33aac750a0907802fb2fc9837ef09ac2e40540bb
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/handlers/polymorphic_belongs_to_handler_spec.rb
@@ -0,0 +1,79 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Handlers::PolymorphicBelongsToHandler, feature_category: :database do
+  let(:model) { Note }
+  let(:association_name) { 'noteable' }
+  let(:reflection) { model.reflections[association_name.to_s] }
+  let(:handler) { described_class.new(model, association_name, reflection) }
+
+  describe '#polymorphic?' do
+    it 'returns true' do
+      expect(handler.polymorphic?).to be true
+    end
+  end
+
+  describe '#build_relationship' do
+    it 'builds a polymorphic belongs_to relationship' do
+      relationship = handler.build_relationship
+
+      expect(relationship).to include(
+        source_table: 'notes',
+        target_table: nil,
+        relationship_type: 'many_to_one',
+        polymorphic: true,
+        polymorphic_type_column: 'noteable_type',
+        polymorphic_name: 'noteable'
+      )
+      expect(relationship[:reverse_association]).to include(
+        name: 'noteable',
+        type: 'belongs_to',
+        model: 'Note'
+      )
+      expect(relationship[:has_ar_reverse]).to be true
+    end
+  end
+
+  describe '#build_relationship_with_target' do
+    it 'builds a specific relationship for a target table' do
+      relationship = handler.build_relationship_with_target('issues')
+
+      expect(relationship).to include(
+        source_table: 'issues',
+        target_table: 'notes',
+        relationship_type: 'one_to_many',
+        polymorphic: true,
+        polymorphic_type_column: 'noteable_type',
+        polymorphic_name: 'noteable'
+      )
+      expect(relationship[:forward_association]).to include(
+        name: 'noteable',
+        type: 'has_many',
+        model: 'Issue'
+      )
+      expect(relationship[:has_ar_forward]).to be true
+    end
+  end
+
+  describe '#discover_targets' do
+    it 'finds models with matching has_many/has_one as: associations' do
+      # Mock models that would have has_many :notes, as: :noteable
+      mock_models = [
+        double('Issue', reflections: {
+          'notes' => double('reflection', macro: :has_many, options: { as: :noteable })
+        }, table_name: 'issues'),
+        double('MergeRequest', reflections: {
+          'notes' => double('reflection', macro: :has_many, options: { as: :noteable })
+        }, table_name: 'merge_requests'),
+        double('Project', reflections: {
+          'issues' => double('reflection', macro: :has_many, options: {})
+        }, table_name: 'projects')
+      ]
+
+      targets = handler.discover_targets(mock_models)
+
+      expect(targets).to contain_exactly('issues', 'merge_requests')
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/handlers/polymorphic_has_association_handler_spec.rb b/spec/lib/gitlab/reflections/relationships/handlers/polymorphic_has_association_handler_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..6574b1eea45fb7c72d408f89d4a1760ba3150c18
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/handlers/polymorphic_has_association_handler_spec.rb
@@ -0,0 +1,79 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Handlers::PolymorphicHasAssociationHandler, feature_category: :database do
+  let(:model) { Issue }
+  let(:association_name) { 'todos' }
+  let(:reflection) { model.reflections[association_name.to_s] }
+  let(:handler) { described_class.new(model, association_name, reflection) }
+
+  describe '#polymorphic?' do
+    it 'returns true' do
+      expect(handler.polymorphic?).to be true
+    end
+  end
+
+  describe '#build_relationship' do
+    it 'builds a polymorphic has_many relationship' do
+      relationship = handler.build_relationship
+
+      expect(relationship).to include(
+        source_table: 'issues',
+        target_table: nil,
+        relationship_type: 'one_to_many',
+        polymorphic: true,
+        polymorphic_type_column: 'target_type',
+        polymorphic_name: 'target'
+      )
+      expect(relationship[:forward_association]).to include(
+        name: 'todos',
+        type: 'has_many',
+        model: 'Issue'
+      )
+      expect(relationship[:has_ar_forward]).to be true
+    end
+  end
+
+  describe '#build_relationship_with_target' do
+    it 'builds a specific relationship for a target table' do
+      relationship = handler.build_relationship_with_target('todos')
+
+      expect(relationship).to include(
+        source_table: 'issues',
+        target_table: 'todos',
+        relationship_type: 'one_to_many',
+        polymorphic: true,
+        polymorphic_type_column: 'target_type',
+        polymorphic_name: 'target'
+      )
+      expect(relationship[:forward_association]).to include(
+        name: 'todos',
+        type: 'has_many',
+        model: 'Issue'
+      )
+      expect(relationship[:has_ar_forward]).to be true
+    end
+  end
+
+  describe '#discover_targets' do
+    it 'finds models with matching belongs_to polymorphic associations' do
+      # Mock models that would have belongs_to :target, polymorphic: true
+      mock_models = [
+        double('Todo', reflections: {
+          'target' => double('reflection', macro: :belongs_to, polymorphic?: true)
+        }, table_name: 'todos'),
+        double('Note', reflections: {
+          'noteable' => double('reflection', macro: :belongs_to, polymorphic?: true)
+        }, table_name: 'notes'),
+        double('Project', reflections: {
+          'namespace' => double('reflection', macro: :belongs_to, polymorphic?: false)
+        }, table_name: 'projects')
+      ]
+
+      targets = handler.discover_targets(mock_models)
+
+      expect(targets).to contain_exactly('todos')
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/transformers/expand_polymorphic_spec.rb b/spec/lib/gitlab/reflections/relationships/transformers/expand_polymorphic_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..920b46c71a8ecb7ff3527294f38b378d5d0d8ce5
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/transformers/expand_polymorphic_spec.rb
@@ -0,0 +1,136 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Transformers::ExpandPolymorphic, feature_category: :database do
+  describe '.call' do
+    let(:polymorphic_relationship) do
+      {
+        'source_table' => 'notes',
+        'target_table' => nil, # Polymorphic, no specific target
+        'foreign_key' => 'noteable_id',
+        'foreign_key_table' => 'notes',
+        'source_columns' => %w[noteable_id noteable_type],
+        'target_columns' => ['id', nil], # Type value will be filled in
+        'relationship_type' => 'belongs_to',
+        'polymorphic' => true,
+        'polymorphic_name' => 'noteable',
+        'polymorphic_type_column' => 'noteable_type'
+      }
+    end
+
+    let(:concrete_relationships) do
+      [
+        {
+          'source_table' => 'notes',
+          'target_table' => 'issues',
+          'foreign_key' => 'noteable_id',
+          'polymorphic_name' => 'noteable',
+          'polymorphic' => true
+        },
+        {
+          'source_table' => 'notes',
+          'target_table' => 'merge_requests',
+          'foreign_key' => 'noteable_id',
+          'polymorphic_name' => 'noteable',
+          'polymorphic' => true
+        }
+      ]
+    end
+
+    let(:relationships) { [polymorphic_relationship] + concrete_relationships }
+
+    subject { described_class.call(relationships) }
+
+    it 'expands polymorphic relationships into concrete has_many relationships with conditions' do
+      result = subject
+
+      # Should have 2 concrete relationships (issues and merge_requests) plus the 2 already concrete ones
+      expect(result.length).to eq(4)
+
+      # Find the expanded relationships
+      issue_relationship = result.find { |r| r['target_table'] == 'issues' && r['conditions'] }
+      merge_request_relationship = result.find { |r| r['target_table'] == 'merge_requests' && r['conditions'] }
+
+      expect(issue_relationship).to be_present
+      expect(merge_request_relationship).to be_present
+
+      # Check issue relationship
+      expect(issue_relationship).to include(
+        'source_table' => 'notes',
+        'target_table' => 'issues',
+        'relationship_type' => 'one_to_many',
+        'polymorphic' => false,
+        'polymorphic_type_column' => 'noteable_type',
+        'polymorphic_type_value' => 'Issue',
+        'conditions' => { 'noteable_type' => 'Issue' }
+      )
+
+      # Check merge request relationship
+      expect(merge_request_relationship).to include(
+        'source_table' => 'notes',
+        'target_table' => 'merge_requests',
+        'relationship_type' => 'one_to_many',
+        'polymorphic' => false,
+        'polymorphic_type_column' => 'noteable_type',
+        'polymorphic_type_value' => 'MergeRequest',
+        'conditions' => { 'noteable_type' => 'MergeRequest' }
+      )
+
+      # Check that source_columns and target_columns include the type column/value
+      expect(issue_relationship['source_columns']).to include('noteable_type')
+      expect(issue_relationship['target_columns']).to include('Issue')
+
+      expect(merge_request_relationship['source_columns']).to include('noteable_type')
+      expect(merge_request_relationship['target_columns']).to include('MergeRequest')
+    end
+
+    it 'preserves non-polymorphic relationships unchanged' do
+      non_polymorphic = {
+        'source_table' => 'users',
+        'target_table' => 'projects',
+        'foreign_key' => 'creator_id',
+        'polymorphic' => false
+      }
+
+      result = described_class.call([non_polymorphic])
+
+      expect(result.length).to eq(1)
+      expect(result.first).to include(non_polymorphic)
+      expect(result.first['conditions']).to be_nil
+    end
+
+    it 'handles concrete polymorphic relationships (those with target_table)' do
+      concrete_polymorphic = {
+        'source_table' => 'notes',
+        'target_table' => 'issues', # Already has target_table - this is a concrete polymorphic relationship
+        'polymorphic' => true,
+        'polymorphic_name' => 'noteable'
+      }
+
+      result = described_class.call([concrete_polymorphic])
+
+      expect(result.length).to eq(1)
+      expect(result.first).to include(concrete_polymorphic)
+      # Should not add conditions to concrete polymorphic relationships
+      expect(result.first['conditions']).to be_nil
+    end
+
+    context 'when polymorphic_type_column is not provided' do
+      let(:polymorphic_relationship_without_type_column) do
+        polymorphic_relationship.except('polymorphic_type_column')
+      end
+
+      let(:relationships) { [polymorphic_relationship_without_type_column] + concrete_relationships }
+
+      it 'infers the type column from the foreign key' do
+        result = subject
+
+        issue_relationship = result.find { |r| r['target_table'] == 'issues' && r['conditions'] }
+
+        expect(issue_relationship['polymorphic_type_column']).to eq('noteable_type')
+        expect(issue_relationship['conditions']).to eq({ 'noteable_type' => 'Issue' })
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/transformers/filter_through_relationships_spec.rb b/spec/lib/gitlab/reflections/relationships/transformers/filter_through_relationships_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..1fc3079fb0c4293d2ac50d61c47ac5c8fe401682
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/transformers/filter_through_relationships_spec.rb
@@ -0,0 +1,138 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Transformers::FilterThroughRelationships do
+  describe '.call' do
+    let(:relationships) do
+      [
+        {
+          source_table: 'users',
+          target_table: 'projects',
+          source_columns: ['id'],
+          target_columns: ['user_id'],
+          is_through_association: false
+        },
+        {
+          source_table: 'users',
+          target_table: 'groups',
+          source_columns: ['id'],
+          target_columns: ['user_id'],
+          through_table: 'members',
+          is_through_association: true
+        },
+        {
+          source_table: 'projects',
+          target_table: 'labels',
+          source_columns: ['id'],
+          target_columns: ['project_id'],
+          through_table: 'project_labels'
+        },
+        {
+          source_table: 'issues',
+          target_table: 'users',
+          source_columns: ['id'],
+          target_columns: ['issue_id'],
+          is_through_association: true
+        },
+        {
+          source_table: 'merge_requests',
+          target_table: 'users',
+          source_columns: ['id'],
+          target_columns: ['merge_request_id']
+        }
+      ]
+    end
+
+    it 'filters out relationships marked as through associations' do
+      result = described_class.call(relationships)
+
+      expect(result).to contain_exactly(
+        {
+          source_table: 'users',
+          target_table: 'projects',
+          source_columns: ['id'],
+          target_columns: ['user_id'],
+          is_through_association: false
+        },
+        {
+          source_table: 'merge_requests',
+          target_table: 'users',
+          source_columns: ['id'],
+          target_columns: ['merge_request_id']
+        }
+      )
+    end
+
+    it 'filters out relationships with through_table' do
+      relationships_with_through_table = [
+        {
+          source_table: 'users',
+          target_table: 'projects',
+          source_columns: ['id'],
+          target_columns: ['user_id']
+        },
+        {
+          source_table: 'projects',
+          target_table: 'labels',
+          source_columns: ['id'],
+          target_columns: ['project_id'],
+          through_table: 'project_labels'
+        }
+      ]
+
+      result = described_class.call(relationships_with_through_table)
+
+      expect(result).to contain_exactly(
+        {
+          source_table: 'users',
+          target_table: 'projects',
+          source_columns: ['id'],
+          target_columns: ['user_id']
+        }
+      )
+    end
+
+    it 'handles relationships with through_relationship? method' do
+      relationship_object = double('relationship')
+      allow(relationship_object).to receive(:respond_to?).with(:through_relationship?).and_return(true)
+      allow(relationship_object).to receive(:through_relationship?).and_return(true)
+      allow(relationship_object).to receive(:[]).with(:is_through_association).and_return(nil)
+      allow(relationship_object).to receive(:[]).with(:through_table).and_return(nil)
+
+      normal_relationship = {
+        source_table: 'users',
+        target_table: 'projects',
+        source_columns: ['id'],
+        target_columns: ['user_id']
+      }
+
+      relationships_with_object = [normal_relationship, relationship_object]
+
+      result = described_class.call(relationships_with_object)
+
+      expect(result).to contain_exactly(normal_relationship)
+    end
+
+    it 'keeps all relationships when none are through relationships' do
+      direct_relationships = [
+        {
+          source_table: 'users',
+          target_table: 'projects',
+          source_columns: ['id'],
+          target_columns: ['user_id']
+        },
+        {
+          source_table: 'projects',
+          target_table: 'issues',
+          source_columns: ['id'],
+          target_columns: ['project_id']
+        }
+      ]
+
+      result = described_class.call(direct_relationships)
+
+      expect(result).to eq(direct_relationships)
+    end
+  end
+end
diff --git a/spec/lib/gitlab/reflections/relationships/transformers/merge_data_sources_spec.rb b/spec/lib/gitlab/reflections/relationships/transformers/merge_data_sources_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..037cc9f0cdaaae387b3512f5d7d20fa7a81ed3b9
--- /dev/null
+++ b/spec/lib/gitlab/reflections/relationships/transformers/merge_data_sources_spec.rb
@@ -0,0 +1,202 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Reflections::Relationships::Transformers::MergeDataSources, feature_category: :database do
+  describe '.call' do
+    subject { described_class.call(relationships) }
+
+    context 'with identical relationships from different data sources' do
+      let(:relationships) do
+        [
+          {
+            'source_table' => 'users',
+            'target_table' => 'projects',
+            'source_columns' => ['id'],
+            'target_columns' => ['user_id'],
+            'data_sources' => ['active_record'],
+            'has_ar_forward' => true,
+            'has_ar_reverse' => false,
+            'has_fk_constraint' => false
+          },
+          {
+            'source_table' => 'users',
+            'target_table' => 'projects',
+            'source_columns' => ['id'],
+            'target_columns' => ['user_id'],
+            'data_sources' => ['foreign_key'],
+            'has_ar_forward' => false,
+            'has_ar_reverse' => false,
+            'has_fk_constraint' => true,
+            'fk_metadata' => { 'constraint_name' => 'fk_projects_user_id' }
+          }
+        ]
+      end
+
+      it 'merges relationships with the same key' do
+        result = subject
+
+        expect(result.length).to eq(1)
+        
+        merged = result.first
+        expect(merged['source_table']).to eq('users')
+        expect(merged['target_table']).to eq('projects')
+        expect(merged['source_columns']).to eq(['id'])
+        expect(merged['target_columns']).to eq(['user_id'])
+        expect(merged['data_sources']).to contain_exactly('active_record', 'foreign_key')
+        expect(merged['has_ar_forward']).to be true
+        expect(merged['has_ar_reverse']).to be false
+        expect(merged['has_fk_constraint']).to be true
+        expect(merged['fk_metadata']).to eq({ 'constraint_name' => 'fk_projects_user_id' })
+      end
+    end
+
+    context 'with different relationships' do
+      let(:relationships) do
+        [
+          {
+            'source_table' => 'users',
+            'target_table' => 'projects',
+            'source_columns' => ['id'],
+            'target_columns' => ['user_id'],
+            'data_sources' => ['active_record']
+          },
+          {
+            'source_table' => 'users',
+            'target_table' => 'issues',
+            'source_columns' => ['id'],
+            'target_columns' => ['author_id'],
+            'data_sources' => ['active_record']
+          }
+        ]
+      end
+
+      it 'keeps different relationships separate' do
+        result = subject
+
+        expect(result.length).to eq(2)
+        
+        user_project = result.find { |r| r['target_table'] == 'projects' }
+        user_issue = result.find { |r| r['target_table'] == 'issues' }
+        
+        expect(user_project).to be_present
+        expect(user_issue).to be_present
+      end
+    end
+
+    context 'with relationships having different column orders' do
+      let(:relationships) do
+        [
+          {
+            'source_table' => 'merge_requests',
+            'target_table' => 'projects',
+            'source_columns' => ['target_project_id', 'source_project_id'],
+            'target_columns' => ['id', 'id'],
+            'data_sources' => ['active_record']
+          },
+          {
+            'source_table' => 'merge_requests',
+            'target_table' => 'projects',
+            'source_columns' => ['source_project_id', 'target_project_id'],
+            'target_columns' => ['id', 'id'],
+            'data_sources' => ['foreign_key']
+          }
+        ]
+      end
+
+      it 'merges relationships with same columns regardless of order' do
+        result = subject
+
+        expect(result.length).to eq(1)
+        
+        merged = result.first
+        expect(merged['data_sources']).to contain_exactly('active_record', 'foreign_key')
+      end
+    end
+
+    context 'with single relationship' do
+      let(:relationships) do
+        [
+          {
+            'source_table' => 'users',
+            'target_table' => 'projects',
+            'source_columns' => ['id'],
+            'target_columns' => ['user_id'],
+            'data_sources' => ['active_record']
+          }
+        ]
+      end
+
+      it 'returns the relationship unchanged' do
+        result = subject
+
+        expect(result.length).to eq(1)
+        expect(result.first).to eq(relationships.first)
+      end
+    end
+
+    context 'with nil and empty data_sources' do
+      let(:relationships) do
+        [
+          {
+            'source_table' => 'users',
+            'target_table' => 'projects',
+            'source_columns' => ['id'],
+            'target_columns' => ['user_id'],
+            'data_sources' => nil,
+            'has_ar_forward' => true
+          },
+          {
+            'source_table' => 'users',
+            'target_table' => 'projects',
+            'source_columns' => ['id'],
+            'target_columns' => ['user_id'],
+            'data_sources' => [],
+            'has_fk_constraint' => true
+          }
+        ]
+      end
+
+      it 'handles nil and empty data_sources correctly' do
+        result = subject
+
+        expect(result.length).to eq(1)
+        
+        merged = result.first
+        expect(merged['data_sources']).to eq([])
+        expect(merged['has_ar_forward']).to be true
+        expect(merged['has_fk_constraint']).to be true
+      end
+    end
+
+    context 'with string vs array columns' do
+      let(:relationships) do
+        [
+          {
+            'source_table' => 'users',
+            'target_table' => 'projects',
+            'source_columns' => 'id',
+            'target_columns' => 'user_id',
+            'data_sources' => ['active_record']
+          },
+          {
+            'source_table' => 'users',
+            'target_table' => 'projects',
+            'source_columns' => ['id'],
+            'target_columns' => ['user_id'],
+            'data_sources' => ['foreign_key']
+          }
+        ]
+      end
+
+      it 'treats string and array columns as equivalent' do
+        result = subject
+
+        expect(result.length).to eq(1)
+        
+        merged = result.first
+        expect(merged['data_sources']).to contain_exactly('active_record', 'foreign_key')
+      end
+    end
+  end
+end
\ No newline at end of file