Extract distinct_relation_for_primary_key on connection

It will allow adapter specific extension (single query by subquery,
distinct on, etc) in a future.

activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb

@@ -1256,6 +1256,25 @@ def columns_for_distinct(columns, orders) # :nodoc:
         columns
       end
 
+      def distinct_relation_for_primary_key(relation) # :nodoc:
+        values = columns_for_distinct(
+          visitor.compile(relation.table[relation.primary_key]),
+          relation.order_values
+        )
+
+        limited = relation.reselect(values).distinct!
+        limited_ids = select_rows(limited.arel, "SQL").map(&:last)
+
+        if limited_ids.empty?
+          relation.none!
+        else
+          relation.where!(relation.primary_key => limited_ids)
+        end
+
+        relation.limit_value = relation.offset_value = nil
+        relation
+      end
+
       # Adds timestamps (+created_at+ and +updated_at+) columns to +table_name+.
       # Additional options (like +:null+) are forwarded to #add_column.
       #

activerecord/lib/active_record/relation/finder_methods.rb

@@ -426,7 +426,7 @@ def apply_join_dependency(eager_loading: group_values.empty?)
         )
         relation = except(:includes, :eager_load, :preload).joins!(join_dependency)
 
-        if eager_loading && !(
+        if eager_loading && has_limit_or_offset? && !(
             using_limitable_reflections?(join_dependency.reflections) &&
             using_limitable_reflections?(
               construct_join_dependency(
@@ -436,11 +436,9 @@ def apply_join_dependency(eager_loading: group_values.empty?)
               ).reflections
             )
         )
-          if has_limit_or_offset?
-            limited_ids = limited_ids_for(relation)
-            limited_ids.empty? ? relation.none! : relation.where!(primary_key => limited_ids)
+          relation = skip_query_cache_if_necessary do
+            klass.connection.distinct_relation_for_primary_key(relation)
           end
-          relation.limit_value = relation.offset_value = nil
         end
 
         if block_given?
@@ -450,18 +448,6 @@ def apply_join_dependency(eager_loading: group_values.empty?)
         end
       end
 
-      def limited_ids_for(relation)
-        values = @klass.connection.columns_for_distinct(
-          connection.visitor.compile(table[primary_key]),
-          relation.order_values
-        )
-
-        relation = relation.except(:select).select(values).distinct!
-
-        id_rows = skip_query_cache_if_necessary { @klass.connection.select_rows(relation.arel, "SQL") }
-        id_rows.map(&:last)
-      end
-
       def using_limitable_reflections?(reflections)
         reflections.none?(&:collection?)
       end