diff --git a/.gitignore b/.gitignore
index 663f28c5..56bc0efb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -19,8 +19,7 @@ test/version_tmp
 tmp
 coverage
 test/log
-test_db
-test_db-journal
+test_db*
 .idea
 *.iml
 *.override.yml
diff --git a/lib/jsonapi-resources.rb b/lib/jsonapi-resources.rb
index e26af1aa..eab7c7c6 100644
--- a/lib/jsonapi-resources.rb
+++ b/lib/jsonapi-resources.rb
@@ -5,6 +5,7 @@
 require 'jsonapi/naive_cache'
 require 'jsonapi/compiled_json'
 require 'jsonapi/relation_retrieval'
+require 'jsonapi/active_relation_retrieval/find_related_through_primary'
 require 'jsonapi/active_relation_retrieval'
 require 'jsonapi/active_relation_retrieval_v09'
 require 'jsonapi/active_relation_retrieval_v10'
@@ -41,8 +42,8 @@
 require 'jsonapi/callbacks'
 require 'jsonapi/link_builder'
 require 'jsonapi/active_relation/adapters/join_left_active_record_adapter'
-require 'jsonapi/active_relation/join_manager'
-require 'jsonapi/active_relation/join_manager_v10'
+require 'jsonapi/active_relation/join_manager_through_inverse'
+require 'jsonapi/active_relation/join_manager_through_primary'
 require 'jsonapi/resource_identity'
 require 'jsonapi/resource_fragment'
 require 'jsonapi/resource_tree'
diff --git a/lib/jsonapi/active_relation/join_manager.rb b/lib/jsonapi/active_relation/join_manager_through_inverse.rb
similarity index 99%
rename from lib/jsonapi/active_relation/join_manager.rb
rename to lib/jsonapi/active_relation/join_manager_through_inverse.rb
index 775831ee..eb9b27b0 100644
--- a/lib/jsonapi/active_relation/join_manager.rb
+++ b/lib/jsonapi/active_relation/join_manager_through_inverse.rb
@@ -5,7 +5,7 @@ module ActiveRelation
 
   # Stores relationship paths starting from the resource_klass, consolidating duplicate paths from
   # relationships, filters and sorts. When joins are made the table aliases are tracked in join_details
-  class JoinManager
+  class JoinManagerThroughInverse
       attr_reader :resource_klass,
                   :source_relationship,
                   :resource_join_tree,
@@ -77,7 +77,7 @@ def join_details_by_relationship(relationship)
         @join_details[segment]
       end
 
-      def self.get_join_arel_node(records, relationship, join_type, options = {})
+      def self.get_join_arel_node(records, relationship, join_type, options)
         init_join_sources = records.arel.join_sources
         init_join_sources_length = init_join_sources.length
 
diff --git a/lib/jsonapi/active_relation/join_manager_v10.rb b/lib/jsonapi/active_relation/join_manager_through_primary.rb
similarity index 99%
rename from lib/jsonapi/active_relation/join_manager_v10.rb
rename to lib/jsonapi/active_relation/join_manager_through_primary.rb
index e8948013..7ee5127c 100644
--- a/lib/jsonapi/active_relation/join_manager_v10.rb
+++ b/lib/jsonapi/active_relation/join_manager_through_primary.rb
@@ -5,7 +5,7 @@ module ActiveRelation
 
     # Stores relationship paths starting from the resource_klass, consolidating duplicate paths from
     # relationships, filters and sorts. When joins are made the table aliases are tracked in join_details
-    class JoinManagerV10
+    class JoinManagerThroughPrimary
       attr_reader :resource_klass,
                   :source_relationship,
                   :resource_join_tree,
@@ -72,7 +72,7 @@ def join_details_by_relationship(relationship)
         @join_details[segment]
       end
 
-      def self.get_join_arel_node(records, relationship, join_type, options = {})
+      def self.get_join_arel_node(records, relationship, join_type, options)
         init_join_sources = records.arel.join_sources
         init_join_sources_length = init_join_sources.length
 
diff --git a/lib/jsonapi/active_relation_retrieval.rb b/lib/jsonapi/active_relation_retrieval.rb
index 88f886c8..ee1d92a6 100644
--- a/lib/jsonapi/active_relation_retrieval.rb
+++ b/lib/jsonapi/active_relation_retrieval.rb
@@ -3,12 +3,23 @@
 module JSONAPI
   module ActiveRelationRetrieval
     include ::JSONAPI::RelationRetrieval
+    include ::JSONAPI::ActiveRelationRetrieval::FindRelatedThroughPrimary
 
-    def find_related_ids(relationship, options = {})
+    def find_related_ids(relationship, options)
       self.class.find_related_fragments(self, relationship, options).keys.collect { |rid| rid.id }
     end
 
     module ClassMethods
+      include JSONAPI::ActiveRelationRetrieval::FindRelatedThroughPrimary::ClassMethods
+
+      def default_find_related_through(polymorphic = false)
+        if polymorphic
+          JSONAPI.configuration.default_find_related_through_polymorphic
+        else
+          JSONAPI.configuration.default_find_related_through
+        end
+      end
+
       # Finds Resources using the `filters`. Pagination and sort options are used when provided
       #
       # @param filters [Hash] the filters hash
@@ -17,12 +28,12 @@ module ClassMethods
       # @option options [Hash] :include_directives The `include_directives`
       #
       # @return [Array<Resource>] the Resource instances matching the filters, sorting and pagination rules.
-      def find(filters, options = {})
+      def find(filters, options)
         sort_criteria = options.fetch(:sort_criteria) { [] }
 
-        join_manager = ActiveRelation::JoinManager.new(resource_klass: self,
-                                                       filters: filters,
-                                                       sort_criteria: sort_criteria)
+        join_manager = ActiveRelation::JoinManagerThroughInverse.new(resource_klass: self,
+                                                                     filters: filters,
+                                                                     sort_criteria: sort_criteria)
 
         paginator = options[:paginator]
 
@@ -41,9 +52,9 @@ def find(filters, options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [Integer] the count
-      def count(filters, options = {})
-        join_manager = ActiveRelation::JoinManager.new(resource_klass: self,
-                                                       filters: filters)
+      def count(filters, options)
+        join_manager = ActiveRelation::JoinManagerThroughInverse.new(resource_klass: self,
+                                                                     filters: filters)
 
         records = apply_request_settings_to_records(records: records(options),
                                filters: filters,
@@ -57,7 +68,7 @@ def count(filters, options = {})
       #
       # @param key the primary key of the resource to find
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_by_key(key, options = {})
+      def find_by_key(key, options)
         record = find_record_by_key(key, options)
         resource_for(record, options[:context])
       end
@@ -66,7 +77,7 @@ def find_by_key(key, options = {})
       #
       # @param keys [Array<key>] Array of primary keys to find resources for
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_by_keys(keys, options = {})
+      def find_by_keys(keys, options)
         records = find_records_by_keys(keys, options)
         resources_for(records, options[:context])
       end
@@ -76,7 +87,7 @@ def find_by_keys(keys, options = {})
       #
       # @param keys [Array<key>] Array of primary keys to find resources for
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_to_populate_by_keys(keys, options = {})
+      def find_to_populate_by_keys(keys, options)
         records = records_for_populate(options).where(_primary_key => keys)
         resources_for(records, options[:context])
       end
@@ -93,7 +104,7 @@ def find_to_populate_by_keys(keys, options = {})
       # @return [Hash{ResourceIdentity => {identity: => ResourceIdentity, cache: cache_field}]
       #    the ResourceInstances matching the filters, sorting, and pagination rules along with any request
       #    additional_field values
-      def find_fragments(filters, options = {})
+      def find_fragments(filters, options)
         include_directives = options.fetch(:include_directives, {})
         resource_klass = self
 
@@ -103,11 +114,11 @@ def find_fragments(filters, options = {})
 
         sort_criteria = options.fetch(:sort_criteria) { [] }
 
-        join_manager = ActiveRelation::JoinManager.new(resource_klass: resource_klass,
-                                                       source_relationship: nil,
-                                                       relationships: linkage_relationships.collect(&:name),
-                                                       sort_criteria: sort_criteria,
-                                                       filters: filters)
+        join_manager = ActiveRelation::JoinManagerThroughInverse.new(resource_klass: resource_klass,
+                                                                     source_relationship: nil,
+                                                                     relationships: linkage_relationships.collect(&:name),
+                                                                     sort_criteria: sort_criteria,
+                                                                     filters: filters)
 
         paginator = options[:paginator]
 
@@ -119,6 +130,11 @@ def find_fragments(filters, options = {})
                                options: options)
 
         if options[:cache]
+          # When using caching the a two step process is used. First the records ids are retrieved and then the
+          # records are retrieved using the ids. Then the ids are used to query the database again to get the
+          # cache misses. In the second phase the records are not sorted or paginated and the `records_for_populate`
+          # method is used to ensure any dependent includes or custom database fields are calculated.
+
           # This alias is going to be resolve down to the model's table name and will not actually be an alias
           resource_table_alias = resource_klass._table_name
 
@@ -189,6 +205,10 @@ def find_fragments(filters, options = {})
             warn "Performance issue detected: `#{self.name.to_s}.records` returned non-normalized results in `#{self.name.to_s}.find_fragments`."
           end
         else
+          # When not using caching resources can be generated after querying. The `records_for_populate`
+          # method is merged in to ensure any dependent includes or custom database fields are calculated.
+          records = records.merge(records_for_populate(options))
+
           linkage_fields = []
 
           linkage_relationships.each do |linkage_relationship|
@@ -259,51 +279,75 @@ def find_fragments(filters, options = {})
       # @return [Hash{ResourceIdentity => {identity: => ResourceIdentity, cache: cache_field, related: {relationship_name: [] }}}]
       #    the ResourceInstances matching the filters, sorting, and pagination rules along with any request
       #    additional_field values
-      def find_related_fragments(source_fragment, relationship, options = {})
-        if relationship.polymorphic? # && relationship.foreign_key_on == :self
-          source_resource_klasses = if relationship.foreign_key_on == :self
-                                      relationship.polymorphic_types.collect do |polymorphic_type|
-                                        resource_klass_for(polymorphic_type)
+      def find_related_fragments(source_fragment, relationship, options)
+        case relationship.find_related_through
+        when :primary
+          if relationship.polymorphic?
+            find_related_polymorphic_fragments_through_primary([source_fragment], relationship, options, false)
+          else
+            find_related_monomorphic_fragments_through_primary([source_fragment], relationship, options, false)
+          end
+        when :inverse
+          if relationship.polymorphic? # && relationship.foreign_key_on == :self
+            source_resource_klasses = if relationship.foreign_key_on == :self
+                                        relationship.polymorphic_types.collect do |polymorphic_type|
+                                          resource_klass_for(polymorphic_type)
+                                        end
+                                      else
+                                        source.collect { |fragment| fragment.identity.resource_klass }.to_set
                                       end
-                                    else
-                                      source.collect { |fragment| fragment.identity.resource_klass }.to_set
-                                    end
 
-          fragments = {}
-          source_resource_klasses.each do |resource_klass|
-            inverse_direct_relationship = _relationship(resource_klass._type.to_s.singularize)
+            fragments = {}
+            source_resource_klasses.each do |resource_klass|
+              inverse_direct_relationship = _relationship(resource_klass._type.to_s.singularize)
 
-            fragments.merge!(resource_klass.find_related_fragments_from_inverse([source_fragment], inverse_direct_relationship, options, false))
+              fragments.merge!(resource_klass.find_related_fragments_through_inverse([source_fragment], inverse_direct_relationship, options, false))
+            end
+            fragments
+          else
+            relationship.resource_klass.find_related_fragments_through_inverse([source_fragment], relationship, options, false)
           end
-          fragments
         else
-          relationship.resource_klass.find_related_fragments_from_inverse([source_fragment], relationship, options, false)
+          raise "Unknown find_related_through: #{relationship.find_related_through}"
+          {}
         end
       end
 
       def find_included_fragments(source_fragments, relationship, options)
-        if relationship.polymorphic? # && relationship.foreign_key_on == :self
-          source_resource_klasses = if relationship.foreign_key_on == :self
-                                      relationship.polymorphic_types.collect do |polymorphic_type|
-                                        resource_klass_for(polymorphic_type)
+        case relationship.find_related_through
+        when :primary
+          if relationship.polymorphic?
+            find_related_polymorphic_fragments_through_primary(source_fragments, relationship, options, true)
+          else
+            find_related_monomorphic_fragments_through_primary(source_fragments, relationship, options, true)
+          end
+        when :inverse
+          if relationship.polymorphic? # && relationship.foreign_key_on == :self
+            source_resource_klasses = if relationship.foreign_key_on == :self
+                                        relationship.polymorphic_types.collect do |polymorphic_type|
+                                          resource_klass_for(polymorphic_type)
+                                        end
+                                      else
+                                        source.collect { |fragment| fragment.identity.resource_klass }.to_set
                                       end
-                                    else
-                                      source_fragments.collect { |fragment| fragment.identity.resource_klass }.to_set
-                                    end
 
-          fragments = {}
-          source_resource_klasses.each do |resource_klass|
-            inverse_direct_relationship = _relationship(resource_klass._type.to_s.singularize)
+            fragments = {}
+            source_resource_klasses.each do |resource_klass|
+              inverse_direct_relationship = _relationship(resource_klass._type.to_s.singularize)
 
-            fragments.merge!(resource_klass.find_related_fragments_from_inverse(source_fragments, inverse_direct_relationship, options, true))
+              fragments.merge!(resource_klass.find_related_fragments_through_inverse(source_fragments, inverse_direct_relationship, options, true))
+            end
+            fragments
+          else
+            relationship.resource_klass.find_related_fragments_through_inverse(source_fragments, relationship, options, true)
           end
-          fragments
         else
-          relationship.resource_klass.find_related_fragments_from_inverse(source_fragments, relationship, options, true)
+          raise "Unknown find_related_through: #{relationship.options[:find_related_through]}"
+          {}
         end
       end
 
-      def find_related_fragments_from_inverse(source, source_relationship, options, connect_source_identity)
+      def find_related_fragments_through_inverse(source, source_relationship, options, connect_source_identity)
         inverse_relationship = source_relationship._inverse_relationship
         return {} if inverse_relationship.blank?
 
@@ -331,11 +375,11 @@ def find_related_fragments_from_inverse(source, source_relationship, options, co
           end
         end
 
-        join_manager = ActiveRelation::JoinManager.new(resource_klass: self,
-                                                       source_relationship: inverse_relationship,
-                                                       relationships: linkage_relationships.collect(&:name),
-                                                       sort_criteria: sort_criteria,
-                                                       filters: filters)
+        join_manager = ActiveRelation::JoinManagerThroughInverse.new(resource_klass: self,
+                                                                     source_relationship: inverse_relationship,
+                                                                     relationships: linkage_relationships.collect(&:name),
+                                                                     sort_criteria: sort_criteria,
+                                                                     filters: filters)
 
         paginator = options[:paginator]
 
@@ -498,11 +542,11 @@ def find_related_fragments_from_inverse(source, source_relationship, options, co
       #
       # @return [Integer] the count
 
-      def count_related(source, relationship, options = {})
-        relationship.resource_klass.count_related_from_inverse(source, relationship, options)
+      def count_related(source, relationship, options)
+        relationship.resource_klass.count_related_through_inverse(source, relationship, options)
       end
 
-      def count_related_from_inverse(source_resource, source_relationship, options = {})
+      def count_related_through_inverse(source_resource, source_relationship, options)
         inverse_relationship = source_relationship._inverse_relationship
         return -1 if inverse_relationship.blank?
 
@@ -511,9 +555,9 @@ def count_related_from_inverse(source_resource, source_relationship, options = {
         filters = options.fetch(:filters, {})
 
         # Joins in this case are related to the related_klass
-        join_manager = ActiveRelation::JoinManager.new(resource_klass: self,
-                                                       source_relationship: inverse_relationship,
-                                                       filters: filters)
+        join_manager = ActiveRelation::JoinManagerThroughInverse.new(resource_klass: self,
+                                                                     source_relationship: inverse_relationship,
+                                                                     filters: filters)
 
         records = apply_request_settings_to_records(records: records(options),
                                                     resource_klass: self,
@@ -542,7 +586,7 @@ def count_related_from_inverse(source_resource, source_relationship, options = {
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_base(_options = {})
+      def records_base(_options)
         _model_class.all
       end
 
@@ -552,7 +596,7 @@ def records_base(_options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records(options = {})
+      def records(options)
         records_base(options)
       end
 
@@ -563,7 +607,7 @@ def records(options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_for_populate(options = {})
+      def records_for_populate(options)
         records_base(options)
       end
 
@@ -572,7 +616,7 @@ def records_for_populate(options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_for_source_to_related(options = {})
+      def records_for_source_to_related(options)
         records_base(options)
       end
 
@@ -637,18 +681,18 @@ def join_relationship(records:, relationship:, resource_type: nil, join_type: :i
 
       # protected
 
-      def find_record_by_key(key, options = {})
+      def find_record_by_key(key, options)
         record = apply_request_settings_to_records(records: records(options), primary_keys: key, options: options).first
         fail JSONAPI::Exceptions::RecordNotFound.new(key) if record.nil?
         record
       end
 
-      def find_records_by_keys(keys, options = {})
+      def find_records_by_keys(keys, options)
         apply_request_settings_to_records(records: records(options), primary_keys: keys, options: options)
       end
 
       def apply_request_settings_to_records(records:,
-                                            join_manager: ActiveRelation::JoinManager.new(resource_klass: self),
+                                            join_manager: ActiveRelation::JoinManagerThroughInverse.new(resource_klass: self),
                                             resource_klass: self,
                                             source_ids: nil,
                                             filters: {},
@@ -839,7 +883,7 @@ def quote(field)
         %{"#{field.to_s}"}
       end
 
-      def apply_filters(records, filters, options = {})
+      def apply_filters(records, filters, options)
         if filters
           filters.each do |filter, value|
             records = apply_filter(records, filter, value, options)
@@ -865,7 +909,7 @@ def get_aliased_field(path_with_field, join_manager)
         concat_table_field(table_alias, field_segment.delegated_field_name)
       end
 
-      def apply_filter(records, filter, value, options = {})
+      def apply_filter(records, filter, value, options)
         strategy = _allowed_filters.fetch(filter.to_sym, Hash.new)[:apply]
 
         if strategy
diff --git a/lib/jsonapi/active_relation_retrieval/find_related_through_primary.rb b/lib/jsonapi/active_relation_retrieval/find_related_through_primary.rb
new file mode 100644
index 00000000..fe3adfbe
--- /dev/null
+++ b/lib/jsonapi/active_relation_retrieval/find_related_through_primary.rb
@@ -0,0 +1,271 @@
+# frozen_string_literal: true
+
+module JSONAPI
+  module ActiveRelationRetrieval
+    module FindRelatedThroughPrimary
+      module ClassMethods
+        def find_related_monomorphic_fragments_through_primary(source_fragments, relationship, options, connect_source_identity)
+          filters = options.fetch(:filters, {})
+          source_ids = source_fragments.collect {|item| item.identity.id}
+
+          include_directives = options.fetch(:include_directives, {})
+          resource_klass = relationship.resource_klass
+          linkage_relationships = resource_klass.to_one_relationships_for_linkage(include_directives[:include_related])
+
+          sort_criteria = []
+          options[:sort_criteria].try(:each) do |sort|
+            field = sort[:field].to_s == 'id' ? resource_klass._primary_key : sort[:field]
+            sort_criteria << { field: field, direction: sort[:direction] }
+          end
+
+          join_manager = ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: self,
+                                                                       source_relationship: relationship,
+                                                                       relationships: linkage_relationships.collect(&:name),
+                                                                       sort_criteria: sort_criteria,
+                                                                       filters: filters)
+
+          paginator = options[:paginator]
+
+          records = apply_request_settings_to_records(records: records_for_source_to_related(options),
+                                                      resource_klass: resource_klass,
+                                                      sort_criteria: sort_criteria,
+                                                      primary_keys: source_ids,
+                                                      paginator: paginator,
+                                                      filters: filters,
+                                                      join_manager: join_manager,
+                                                      options: options)
+
+          resource_table_alias = join_manager.join_details_by_relationship(relationship)[:alias]
+
+          pluck_fields = [
+            Arel.sql("#{_table_name}.#{_primary_key} AS \"source_id\""),
+            sql_field_with_alias(resource_table_alias, resource_klass._primary_key)
+          ]
+
+          cache_field = resource_klass.attribute_to_model_field(:_cache_field) if options[:cache]
+          if cache_field
+            pluck_fields << sql_field_with_alias(resource_table_alias, cache_field[:name])
+          end
+
+          linkage_fields = []
+
+          linkage_relationships.each do |linkage_relationship|
+            linkage_relationship_name = linkage_relationship.name
+
+            if linkage_relationship.polymorphic? && linkage_relationship.belongs_to?
+              linkage_relationship.resource_types.each do |resource_type|
+                klass = resource_klass_for(resource_type)
+                linkage_fields << {relationship_name: linkage_relationship_name, resource_klass: klass}
+
+                linkage_table_alias = join_manager.join_details_by_polymorphic_relationship(linkage_relationship, resource_type)[:alias]
+                primary_key = klass._primary_key
+                pluck_fields << sql_field_with_alias(linkage_table_alias, primary_key)
+              end
+            else
+              klass = linkage_relationship.resource_klass
+              linkage_fields << {relationship_name: linkage_relationship_name, resource_klass: klass}
+
+              linkage_table_alias = join_manager.join_details_by_relationship(linkage_relationship)[:alias]
+              primary_key = klass._primary_key
+              pluck_fields << sql_field_with_alias(linkage_table_alias, primary_key)
+            end
+          end
+
+          sort_fields = options.dig(:_relation_helper_options, :sort_fields)
+          sort_fields.try(:each) do |field|
+            pluck_fields << Arel.sql(field)
+          end
+
+          fragments = {}
+          rows = records.distinct.pluck(*pluck_fields)
+          rows.each do |row|
+            rid = JSONAPI::ResourceIdentity.new(resource_klass, row[1])
+
+            fragments[rid] ||= JSONAPI::ResourceFragment.new(rid)
+
+            attributes_offset = 2
+
+            if cache_field
+              fragments[rid].cache = cast_to_attribute_type(row[attributes_offset], cache_field[:type])
+              attributes_offset+= 1
+            end
+
+            source_rid = JSONAPI::ResourceIdentity.new(self, row[0])
+
+            fragments[rid].add_related_from(source_rid)
+
+            linkage_fields.each do |linkage_field|
+              fragments[rid].initialize_related(linkage_field[:relationship_name])
+              related_id = row[attributes_offset]
+              if related_id
+                related_rid = JSONAPI::ResourceIdentity.new(linkage_field[:resource_klass], related_id)
+                fragments[rid].add_related_identity(linkage_field[:relationship_name], related_rid)
+              end
+              attributes_offset+= 1
+            end
+
+            if connect_source_identity
+              inverse_relationship = relationship._inverse_relationship
+              fragments[rid].add_related_identity(inverse_relationship.name, source_rid) if inverse_relationship.present?
+            end
+          end
+
+          fragments
+        end
+
+        # Gets resource identities where the related resource is polymorphic and the resource type and id
+        # are stored on the primary resources. Cache fields will always be on the related resources.
+        def find_related_polymorphic_fragments_through_primary(source_fragments, relationship, options, connect_source_identity)
+          filters = options.fetch(:filters, {})
+          source_ids = source_fragments.collect {|item| item.identity.id}
+
+          resource_klass = relationship.resource_klass
+          include_directives = options.fetch(:include_directives, {})
+
+          linkage_relationship_paths = []
+
+          resource_types = relationship.resource_types
+
+          resource_types.each do |resource_type|
+            related_resource_klass = resource_klass_for(resource_type)
+            relationships = related_resource_klass.to_one_relationships_for_linkage(include_directives[:include_related])
+            relationships.each do |r|
+              linkage_relationship_paths << "##{resource_type}.#{r.name}"
+            end
+          end
+
+          join_manager = ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: self,
+                                                                       source_relationship: relationship,
+                                                                       relationships: linkage_relationship_paths,
+                                                                       filters: filters)
+
+          paginator = options[:paginator]
+
+          # Note: We will sort by the source table. Without using unions we can't sort on a polymorphic relationship
+          # in any manner that makes sense
+          records = apply_request_settings_to_records(records: records_for_source_to_related(options),
+                                                      resource_klass: resource_klass,
+                                                      sort_primary: true,
+                                                      primary_keys: source_ids,
+                                                      paginator: paginator,
+                                                      filters: filters,
+                                                      join_manager: join_manager,
+                                                      options: options)
+
+          primary_key = concat_table_field(_table_name, _primary_key)
+          related_key = concat_table_field(_table_name, relationship.foreign_key)
+          related_type = concat_table_field(_table_name, relationship.polymorphic_type)
+
+          pluck_fields = [
+            Arel.sql("#{primary_key} AS #{alias_table_field(_table_name, _primary_key)}"),
+            Arel.sql("#{related_key} AS #{alias_table_field(_table_name, relationship.foreign_key)}"),
+            Arel.sql("#{related_type} AS #{alias_table_field(_table_name, relationship.polymorphic_type)}")
+          ]
+
+          # Get the additional fields from each relation. There's a limitation that the fields must exist in each relation
+
+          relation_positions = {}
+          relation_index = pluck_fields.length
+
+          # Add resource specific fields
+          if resource_types.nil? || resource_types.length == 0
+            # :nocov:
+            warn "No resource types found for polymorphic relationship."
+            # :nocov:
+          else
+            resource_types.try(:each) do |type|
+              related_klass = resource_klass_for(type.to_s)
+
+              cache_field = related_klass.attribute_to_model_field(:_cache_field) if options[:cache]
+
+              table_alias = join_manager.source_join_details(type)[:alias]
+
+              cache_offset = relation_index
+              if cache_field
+                pluck_fields << sql_field_with_alias(table_alias, cache_field[:name])
+                relation_index+= 1
+              end
+
+              relation_positions[type] = {relation_klass: related_klass,
+                                          cache_field: cache_field,
+                                          cache_offset: cache_offset}
+            end
+          end
+
+          # Add to_one linkage fields
+          linkage_fields = []
+          linkage_offset = relation_index
+
+          linkage_relationship_paths.each do |linkage_relationship_path|
+            path = JSONAPI::Path.new(resource_klass: self,
+                                     path_string: "#{relationship.name}#{linkage_relationship_path}",
+                                     ensure_default_field: false)
+
+            linkage_relationship = path.segments[-1].relationship
+
+            if linkage_relationship.polymorphic? && linkage_relationship.belongs_to?
+              linkage_relationship.resource_types.each do |resource_type|
+                klass = resource_klass_for(resource_type)
+                linkage_fields << {relationship: linkage_relationship, resource_klass: klass}
+
+                linkage_table_alias = join_manager.join_details_by_polymorphic_relationship(linkage_relationship, resource_type)[:alias]
+                primary_key = klass._primary_key
+                pluck_fields << sql_field_with_alias(linkage_table_alias, primary_key)
+              end
+            else
+              klass = linkage_relationship.resource_klass
+              linkage_fields << {relationship: linkage_relationship, resource_klass: klass}
+
+              linkage_table_alias = join_manager.join_details_by_relationship(linkage_relationship)[:alias]
+              primary_key = klass._primary_key
+              pluck_fields << sql_field_with_alias(linkage_table_alias, primary_key)
+            end
+          end
+
+          rows = records.distinct.pluck(*pluck_fields)
+
+          related_fragments = {}
+
+          rows.each do |row|
+            unless row[1].nil? || row[2].nil?
+              related_klass = resource_klass_for(row[2])
+
+              rid = JSONAPI::ResourceIdentity.new(related_klass, row[1])
+              related_fragments[rid] ||= JSONAPI::ResourceFragment.new(rid)
+
+              source_rid = JSONAPI::ResourceIdentity.new(self, row[0])
+              related_fragments[rid].add_related_from(source_rid)
+
+              if connect_source_identity
+                inverse_relationship = relationship._inverse_relationship
+                related_fragments[rid].add_related_identity(inverse_relationship.name, source_rid) if inverse_relationship.present?
+              end
+
+              relation_position = relation_positions[row[2].underscore.pluralize]
+              model_fields = relation_position[:model_fields]
+              cache_field = relation_position[:cache_field]
+              cache_offset = relation_position[:cache_offset]
+              field_offset = relation_position[:field_offset]
+
+              if cache_field
+                related_fragments[rid].cache = cast_to_attribute_type(row[cache_offset], cache_field[:type])
+              end
+
+              linkage_fields.each_with_index do |linkage_field_details, idx|
+                relationship = linkage_field_details[:relationship]
+                related_fragments[rid].initialize_related(relationship.name)
+                related_id = row[linkage_offset + idx]
+                if related_id
+                  related_rid = JSONAPI::ResourceIdentity.new(linkage_field_details[:resource_klass], related_id)
+                  related_fragments[rid].add_related_identity(relationship.name, related_rid)
+                end
+              end
+            end
+          end
+
+          related_fragments
+        end
+      end
+    end
+  end
+end
diff --git a/lib/jsonapi/active_relation_retrieval_v09.rb b/lib/jsonapi/active_relation_retrieval_v09.rb
index ca72a7b3..fe105a32 100644
--- a/lib/jsonapi/active_relation_retrieval_v09.rb
+++ b/lib/jsonapi/active_relation_retrieval_v09.rb
@@ -4,7 +4,7 @@ module JSONAPI
   module ActiveRelationRetrievalV09
     include ::JSONAPI::RelationRetrieval
 
-    def find_related_ids(relationship, options = {})
+    def find_related_ids(relationship, options)
       self.class.find_related_fragments(self.fragment, relationship, options).keys.collect { |rid| rid.id }
     end
 
@@ -23,7 +23,7 @@ module ClassMethods
       # @option options [Hash] :include_directives The `include_directives`
       #
       # @return [Array<Resource>] the Resource instances matching the filters, sorting and pagination rules.
-      def find(filters, options = {})
+      def find(filters, options)
         context = options[:context]
 
         records = filter_records(records(options), filters, options)
@@ -43,7 +43,7 @@ def find(filters, options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [Integer] the count
-      def count(filters, options = {})
+      def count(filters, options)
         count_records(filter_records(records(options), filters, options))
       end
 
@@ -51,7 +51,7 @@ def count(filters, options = {})
       #
       # @param key the primary key of the resource to find
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_by_key(key, options = {})
+      def find_by_key(key, options)
         context = options[:context]
         records = records(options)
 
@@ -65,7 +65,7 @@ def find_by_key(key, options = {})
       #
       # @param keys [Array<key>] Array of primary keys to find resources for
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_by_keys(keys, options = {})
+      def find_by_keys(keys, options)
         context = options[:context]
         records = records(options)
         records = apply_includes(records, options)
@@ -80,7 +80,7 @@ def find_by_keys(keys, options = {})
       #
       # @param keys [Array<key>] Array of primary keys to find resources for
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_to_populate_by_keys(keys, options = {})
+      def find_to_populate_by_keys(keys, options)
         records = records_for_populate(options).where(_primary_key => keys)
         resources_for(records, options[:context])
       end
@@ -97,15 +97,15 @@ def find_to_populate_by_keys(keys, options = {})
       # @return [Hash{ResourceIdentity => {identity: => ResourceIdentity, cache: cache_field}]
       #    the ResourceInstances matching the filters, sorting, and pagination rules along with any request
       #    additional_field values
-      def find_fragments(filters, options = {})
+      def find_fragments(filters, options)
         context = options[:context]
 
         sort_criteria = options.fetch(:sort_criteria) { [] }
         order_options = construct_order_options(sort_criteria)
 
-        join_manager = ActiveRelation::JoinManager.new(resource_klass: self,
-                                                       filters: filters,
-                                                       sort_criteria: sort_criteria)
+        join_manager = ActiveRelation::JoinManagerThroughInverse.new(resource_klass: self,
+                                                                     filters: filters,
+                                                                     sort_criteria: sort_criteria)
 
         options[:_relation_helper_options] = {
           context: context,
@@ -208,7 +208,7 @@ def find_related_fragments_from_inverse(source, source_relationship, options, co
       #
       # @return [Integer] the count
 
-      def count_related(source, relationship, options = {})
+      def count_related(source, relationship, options)
         opts = options.except(:paginator)
 
         related_resource_records = source.public_send("records_for_#{relationship.name}",
@@ -229,7 +229,7 @@ def count_related(source, relationship, options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_base(_options = {})
+      def records_base(_options)
         _model_class.all
       end
 
@@ -239,7 +239,7 @@ def records_base(_options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records(options = {})
+      def records(options)
         records_base(options)
       end
 
@@ -250,7 +250,7 @@ def records(options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_for_populate(options = {})
+      def records_for_populate(options)
         records_base(options)
       end
 
@@ -259,7 +259,7 @@ def records_for_populate(options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_for_source_to_related(options = {})
+      def records_for_source_to_related(options)
         records_base(options)
       end
 
@@ -481,7 +481,7 @@ def build_to_many(relationship, foreign_key, associated_records_method_name, rel
         end
       end
 
-      def resolve_relationship_names_to_relations(resource_klass, model_includes, options = {})
+      def resolve_relationship_names_to_relations(resource_klass, model_includes, options)
         case model_includes
         when Array
           return model_includes.map do |value|
@@ -501,7 +501,7 @@ def resolve_relationship_names_to_relations(resource_klass, model_includes, opti
         end
       end
 
-      def apply_includes(records, options = {})
+      def apply_includes(records, options)
         include_directives = options[:include_directives]
         if include_directives
           model_includes = resolve_relationship_names_to_relations(self, include_directives.model_includes, options)
@@ -615,7 +615,7 @@ def get_aliased_field(path_with_field, join_manager)
         concat_table_field(table_alias, field_segment.delegated_field_name)
       end
 
-      def apply_filter(records, filter, value, options = {})
+      def apply_filter(records, filter, value, options)
         strategy = _allowed_filters.fetch(filter.to_sym, Hash.new)[:apply]
 
         if strategy
@@ -629,7 +629,7 @@ def apply_filter(records, filter, value, options = {})
         records
       end
 
-      def apply_filters(records, filters, options = {})
+      def apply_filters(records, filters, options)
         # required_includes = []
 
         if filters
@@ -670,7 +670,7 @@ def construct_order_options(sort_params)
         end
       end
 
-      def sort_records(records, order_options, options = {})
+      def sort_records(records, order_options, options)
         apply_sort(records, order_options, options)
       end
 
@@ -679,7 +679,7 @@ def count_records(records)
         records.count(:all)
       end
 
-      def find_count(filters, options = {})
+      def find_count(filters, options)
         count_records(filter_records(records(options), filters, options))
       end
 
diff --git a/lib/jsonapi/active_relation_retrieval_v10.rb b/lib/jsonapi/active_relation_retrieval_v10.rb
index e2ee9b2c..d29e7351 100644
--- a/lib/jsonapi/active_relation_retrieval_v10.rb
+++ b/lib/jsonapi/active_relation_retrieval_v10.rb
@@ -4,11 +4,13 @@ module JSONAPI
   module ActiveRelationRetrievalV10
     include ::JSONAPI::RelationRetrieval
 
-    def find_related_ids(relationship, options = {})
+    def find_related_ids(relationship, options)
       self.class.find_related_fragments(self, relationship, options).keys.collect { |rid| rid.id }
     end
 
     module ClassMethods
+      include JSONAPI::ActiveRelationRetrieval::FindRelatedThroughPrimary::ClassMethods
+
       # Finds Resources using the `filters`. Pagination and sort options are used when provided
       #
       # @param filters [Hash] the filters hash
@@ -17,12 +19,12 @@ module ClassMethods
       # @option options [Hash] :include_directives The `include_directives`
       #
       # @return [Array<Resource>] the Resource instances matching the filters, sorting and pagination rules.
-      def find(filters, options = {})
+      def find(filters, options)
         sort_criteria = options.fetch(:sort_criteria) { [] }
 
-        join_manager = ActiveRelation::JoinManagerV10.new(resource_klass: self,
-                                                       filters: filters,
-                                                       sort_criteria: sort_criteria)
+        join_manager = ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: self,
+                                                                     filters: filters,
+                                                                     sort_criteria: sort_criteria)
 
         paginator = options[:paginator]
 
@@ -41,9 +43,9 @@ def find(filters, options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [Integer] the count
-      def count(filters, options = {})
-        join_manager = ActiveRelation::JoinManagerV10.new(resource_klass: self,
-                                                       filters: filters)
+      def count(filters, options)
+        join_manager = ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: self,
+                                                                     filters: filters)
 
         records = apply_request_settings_to_records(records: records(options),
                                filters: filters,
@@ -57,7 +59,7 @@ def count(filters, options = {})
       #
       # @param key the primary key of the resource to find
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_by_key(key, options = {})
+      def find_by_key(key, options)
         record = find_record_by_key(key, options)
         resource_for(record, options[:context])
       end
@@ -66,7 +68,7 @@ def find_by_key(key, options = {})
       #
       # @param keys [Array<key>] Array of primary keys to find resources for
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_by_keys(keys, options = {})
+      def find_by_keys(keys, options)
         records = find_records_by_keys(keys, options)
         resources_for(records, options[:context])
       end
@@ -76,7 +78,7 @@ def find_by_keys(keys, options = {})
       #
       # @param keys [Array<key>] Array of primary keys to find resources for
       # @option options [Hash] :context The context of the request, set in the controller
-      def find_to_populate_by_keys(keys, options = {})
+      def find_to_populate_by_keys(keys, options)
         records = records_for_populate(options).where(_primary_key => keys)
         resources_for(records, options[:context])
       end
@@ -93,7 +95,7 @@ def find_to_populate_by_keys(keys, options = {})
       # @return [Hash{ResourceIdentity => {identity: => ResourceIdentity, cache: cache_field}]
       #    the ResourceInstances matching the filters, sorting, and pagination rules along with any request
       #    additional_field values
-      def find_fragments(filters, options = {})
+      def find_fragments(filters, options)
         include_directives = options.fetch(:include_directives, {})
         resource_klass = self
 
@@ -103,11 +105,11 @@ def find_fragments(filters, options = {})
 
         sort_criteria = options.fetch(:sort_criteria) { [] }
 
-        join_manager = ActiveRelation::JoinManagerV10.new(resource_klass: resource_klass,
-                                                       source_relationship: nil,
-                                                       relationships: linkage_relationships.collect(&:name),
-                                                       sort_criteria: sort_criteria,
-                                                       filters: filters)
+        join_manager = ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: resource_klass,
+                                                                     source_relationship: nil,
+                                                                     relationships: linkage_relationships.collect(&:name),
+                                                                     sort_criteria: sort_criteria,
+                                                                     filters: filters)
 
         paginator = options[:paginator]
 
@@ -205,7 +207,7 @@ def find_fragments(filters, options = {})
       # @return [Hash{ResourceIdentity => {identity: => ResourceIdentity, cache: cache_field, related: {relationship_name: [] }}}]
       #    the ResourceInstances matching the filters, sorting, and pagination rules along with any request
       #    additional_field values
-      def find_related_fragments(source_fragment, relationship, options = {})
+      def find_related_fragments(source_fragment, relationship, options)
         if relationship.polymorphic?
           find_related_polymorphic_fragments([source_fragment], relationship, options, false)
         else
@@ -228,15 +230,15 @@ def find_included_fragments(source_fragments, relationship, options)
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [Integer] the count
-      def count_related(source_resource, relationship, options = {})
+      def count_related(source_resource, relationship, options)
         related_klass = relationship.resource_klass
 
         filters = options.fetch(:filters, {})
 
         # Joins in this case are related to the related_klass
-        join_manager = ActiveRelation::JoinManagerV10.new(resource_klass: self,
-                                                       source_relationship: relationship,
-                                                       filters: filters)
+        join_manager = ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: self,
+                                                                     source_relationship: relationship,
+                                                                     filters: filters)
 
         records = apply_request_settings_to_records(records: records(options),
                                resource_klass: related_klass,
@@ -265,7 +267,7 @@ def count_related(source_resource, relationship, options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_base(_options = {})
+      def records_base(_options)
         _model_class.all
       end
 
@@ -275,7 +277,7 @@ def records_base(_options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records(options = {})
+      def records(options)
         records_base(options)
       end
 
@@ -286,7 +288,7 @@ def records(options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_for_populate(options = {})
+      def records_for_populate(options)
         records_base(options)
       end
 
@@ -295,7 +297,7 @@ def records_for_populate(options = {})
       # @option options [Hash] :context The context of the request, set in the controller
       #
       # @return [ActiveRecord::Relation]
-      def records_for_source_to_related(options = {})
+      def records_for_source_to_related(options)
         records_base(options)
       end
 
@@ -353,280 +355,18 @@ def join_relationship(records:, relationship:, resource_type: nil, join_type: :i
 
       # protected
 
-      def find_record_by_key(key, options = {})
+      def find_record_by_key(key, options)
         record = apply_request_settings_to_records(records: records(options), primary_keys: key, options: options).first
         fail JSONAPI::Exceptions::RecordNotFound.new(key) if record.nil?
         record
       end
 
-      def find_records_by_keys(keys, options = {})
+      def find_records_by_keys(keys, options)
         apply_request_settings_to_records(records: records(options), primary_keys: keys, options: options)
       end
 
-      def find_related_monomorphic_fragments(source_fragments, relationship, options, connect_source_identity)
-        filters = options.fetch(:filters, {})
-        source_ids = source_fragments.collect {|item| item.identity.id}
-
-        include_directives = options.fetch(:include_directives, {})
-        resource_klass = relationship.resource_klass
-        linkage_relationships = resource_klass.to_one_relationships_for_linkage(include_directives[:include_related])
-
-        sort_criteria = []
-        options[:sort_criteria].try(:each) do |sort|
-          field = sort[:field].to_s == 'id' ? resource_klass._primary_key : sort[:field]
-          sort_criteria << { field: field, direction: sort[:direction] }
-        end
-
-        join_manager = ActiveRelation::JoinManagerV10.new(resource_klass: self,
-                                                       source_relationship: relationship,
-                                                       relationships: linkage_relationships.collect(&:name),
-                                                       sort_criteria: sort_criteria,
-                                                       filters: filters)
-
-        paginator = options[:paginator]
-
-        records = apply_request_settings_to_records(records: records_for_source_to_related(options),
-                               resource_klass: resource_klass,
-                               sort_criteria: sort_criteria,
-                               primary_keys: source_ids,
-                               paginator: paginator,
-                               filters: filters,
-                               join_manager: join_manager,
-                               options: options)
-
-        resource_table_alias = join_manager.join_details_by_relationship(relationship)[:alias]
-
-        pluck_fields = [
-          Arel.sql("#{_table_name}.#{_primary_key} AS \"source_id\""),
-          sql_field_with_alias(resource_table_alias, resource_klass._primary_key)
-        ]
-
-        cache_field = resource_klass.attribute_to_model_field(:_cache_field) if options[:cache]
-        if cache_field
-          pluck_fields << sql_field_with_alias(resource_table_alias, cache_field[:name])
-        end
-
-        linkage_fields = []
-
-        linkage_relationships.each do |linkage_relationship|
-          linkage_relationship_name = linkage_relationship.name
-
-          if linkage_relationship.polymorphic? && linkage_relationship.belongs_to?
-            linkage_relationship.resource_types.each do |resource_type|
-              klass = resource_klass_for(resource_type)
-              linkage_fields << {relationship_name: linkage_relationship_name, resource_klass: klass}
-
-              linkage_table_alias = join_manager.join_details_by_polymorphic_relationship(linkage_relationship, resource_type)[:alias]
-              primary_key = klass._primary_key
-              pluck_fields << sql_field_with_alias(linkage_table_alias, primary_key)
-            end
-          else
-            klass = linkage_relationship.resource_klass
-            linkage_fields << {relationship_name: linkage_relationship_name, resource_klass: klass}
-
-            linkage_table_alias = join_manager.join_details_by_relationship(linkage_relationship)[:alias]
-            primary_key = klass._primary_key
-            pluck_fields << sql_field_with_alias(linkage_table_alias, primary_key)
-          end
-        end
-
-        sort_fields = options.dig(:_relation_helper_options, :sort_fields)
-        sort_fields.try(:each) do |field|
-          pluck_fields << Arel.sql(field)
-        end
-
-        fragments = {}
-        rows = records.distinct.pluck(*pluck_fields)
-        rows.each do |row|
-          rid = JSONAPI::ResourceIdentity.new(resource_klass, row[1])
-
-          fragments[rid] ||= JSONAPI::ResourceFragment.new(rid)
-
-          attributes_offset = 2
-
-          if cache_field
-            fragments[rid].cache = cast_to_attribute_type(row[attributes_offset], cache_field[:type])
-            attributes_offset+= 1
-          end
-
-          source_rid = JSONAPI::ResourceIdentity.new(self, row[0])
-
-          fragments[rid].add_related_from(source_rid)
-
-          linkage_fields.each do |linkage_field|
-            fragments[rid].initialize_related(linkage_field[:relationship_name])
-            related_id = row[attributes_offset]
-            if related_id
-              related_rid = JSONAPI::ResourceIdentity.new(linkage_field[:resource_klass], related_id)
-              fragments[rid].add_related_identity(linkage_field[:relationship_name], related_rid)
-            end
-            attributes_offset+= 1
-          end
-
-          if connect_source_identity
-            inverse_relationship = relationship._inverse_relationship
-            fragments[rid].add_related_identity(inverse_relationship.name, source_rid) if inverse_relationship.present?
-          end
-        end
-
-        fragments
-      end
-
-      # Gets resource identities where the related resource is polymorphic and the resource type and id
-      # are stored on the primary resources. Cache fields will always be on the related resources.
-      def find_related_polymorphic_fragments(source_fragments, relationship, options, connect_source_identity)
-        filters = options.fetch(:filters, {})
-        source_ids = source_fragments.collect {|item| item.identity.id}
-
-        resource_klass = relationship.resource_klass
-        include_directives = options.fetch(:include_directives, {})
-
-        linkage_relationship_paths = []
-
-        resource_types = relationship.resource_types
-
-        resource_types.each do |resource_type|
-          related_resource_klass = resource_klass_for(resource_type)
-          relationships = related_resource_klass.to_one_relationships_for_linkage(include_directives[:include_related])
-          relationships.each do |r|
-            linkage_relationship_paths << "##{resource_type}.#{r.name}"
-          end
-        end
-
-        join_manager = ActiveRelation::JoinManagerV10.new(resource_klass: self,
-                                                       source_relationship: relationship,
-                                                       relationships: linkage_relationship_paths,
-                                                       filters: filters)
-
-        paginator = options[:paginator]
-
-        # Note: We will sort by the source table. Without using unions we can't sort on a polymorphic relationship
-        # in any manner that makes sense
-        records = apply_request_settings_to_records(records: records_for_source_to_related(options),
-                               resource_klass: resource_klass,
-                               sort_primary: true,
-                               primary_keys: source_ids,
-                               paginator: paginator,
-                               filters: filters,
-                               join_manager: join_manager,
-                               options: options)
-
-        primary_key = concat_table_field(_table_name, _primary_key)
-        related_key = concat_table_field(_table_name, relationship.foreign_key)
-        related_type = concat_table_field(_table_name, relationship.polymorphic_type)
-
-        pluck_fields = [
-            Arel.sql("#{primary_key} AS #{alias_table_field(_table_name, _primary_key)}"),
-            Arel.sql("#{related_key} AS #{alias_table_field(_table_name, relationship.foreign_key)}"),
-            Arel.sql("#{related_type} AS #{alias_table_field(_table_name, relationship.polymorphic_type)}")
-        ]
-
-        # Get the additional fields from each relation. There's a limitation that the fields must exist in each relation
-
-        relation_positions = {}
-        relation_index = pluck_fields.length
-
-        # Add resource specific fields
-        if resource_types.nil? || resource_types.length == 0
-          # :nocov:
-          warn "No resource types found for polymorphic relationship."
-          # :nocov:
-        else
-          resource_types.try(:each) do |type|
-            related_klass = resource_klass_for(type.to_s)
-
-            cache_field = related_klass.attribute_to_model_field(:_cache_field) if options[:cache]
-
-            table_alias = join_manager.source_join_details(type)[:alias]
-
-            cache_offset = relation_index
-            if cache_field
-              pluck_fields << sql_field_with_alias(table_alias, cache_field[:name])
-              relation_index+= 1
-            end
-
-            relation_positions[type] = {relation_klass: related_klass,
-                                        cache_field: cache_field,
-                                        cache_offset: cache_offset}
-          end
-        end
-
-        # Add to_one linkage fields
-        linkage_fields = []
-        linkage_offset = relation_index
-
-        linkage_relationship_paths.each do |linkage_relationship_path|
-          path = JSONAPI::Path.new(resource_klass: self,
-                                   path_string: "#{relationship.name}#{linkage_relationship_path}",
-                                   ensure_default_field: false)
-
-          linkage_relationship = path.segments[-1].relationship
-
-          if linkage_relationship.polymorphic? && linkage_relationship.belongs_to?
-            linkage_relationship.resource_types.each do |resource_type|
-              klass = resource_klass_for(resource_type)
-              linkage_fields << {relationship: linkage_relationship, resource_klass: klass}
-
-              linkage_table_alias = join_manager.join_details_by_polymorphic_relationship(linkage_relationship, resource_type)[:alias]
-              primary_key = klass._primary_key
-              pluck_fields << sql_field_with_alias(linkage_table_alias, primary_key)
-            end
-          else
-            klass = linkage_relationship.resource_klass
-            linkage_fields << {relationship: linkage_relationship, resource_klass: klass}
-
-            linkage_table_alias = join_manager.join_details_by_relationship(linkage_relationship)[:alias]
-            primary_key = klass._primary_key
-            pluck_fields << sql_field_with_alias(linkage_table_alias, primary_key)
-          end
-        end
-
-        rows = records.distinct.pluck(*pluck_fields)
-
-        related_fragments = {}
-
-        rows.each do |row|
-          unless row[1].nil? || row[2].nil?
-            related_klass = resource_klass_for(row[2])
-
-            rid = JSONAPI::ResourceIdentity.new(related_klass, row[1])
-            related_fragments[rid] ||= JSONAPI::ResourceFragment.new(rid)
-
-            source_rid = JSONAPI::ResourceIdentity.new(self, row[0])
-            related_fragments[rid].add_related_from(source_rid)
-
-            if connect_source_identity
-              inverse_relationship = relationship._inverse_relationship
-              related_fragments[rid].add_related_identity(inverse_relationship.name, source_rid) if inverse_relationship.present?
-            end
-
-            relation_position = relation_positions[row[2].underscore.pluralize]
-            model_fields = relation_position[:model_fields]
-            cache_field = relation_position[:cache_field]
-            cache_offset = relation_position[:cache_offset]
-            field_offset = relation_position[:field_offset]
-
-            if cache_field
-              related_fragments[rid].cache = cast_to_attribute_type(row[cache_offset], cache_field[:type])
-            end
-
-            linkage_fields.each_with_index do |linkage_field_details, idx|
-              relationship = linkage_field_details[:relationship]
-              related_fragments[rid].initialize_related(relationship.name)
-              related_id = row[linkage_offset + idx]
-              if related_id
-                related_rid = JSONAPI::ResourceIdentity.new(linkage_field_details[:resource_klass], related_id)
-                related_fragments[rid].add_related_identity(relationship.name, related_rid)
-              end
-            end
-          end
-        end
-
-        related_fragments
-      end
-
       def apply_request_settings_to_records(records:,
-                                            join_manager: ActiveRelation::JoinManagerV10.new(resource_klass: self),
+                                            join_manager: ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: self),
                                             resource_klass: self,
                                             filters: {},
                                             primary_keys: nil,
@@ -804,7 +544,7 @@ def quote(field)
         %{"#{field.to_s}"}
       end
 
-      def apply_filters(records, filters, options = {})
+      def apply_filters(records, filters, options)
         if filters
           filters.each do |filter, value|
             records = apply_filter(records, filter, value, options)
@@ -830,7 +570,7 @@ def get_aliased_field(path_with_field, join_manager)
         concat_table_field(table_alias, field_segment.delegated_field_name)
       end
 
-      def apply_filter(records, filter, value, options = {})
+      def apply_filter(records, filter, value, options)
         strategy = _allowed_filters.fetch(filter.to_sym, Hash.new)[:apply]
 
         if strategy
diff --git a/lib/jsonapi/configuration.rb b/lib/jsonapi/configuration.rb
index bfaeb5fd..e75287f3 100644
--- a/lib/jsonapi/configuration.rb
+++ b/lib/jsonapi/configuration.rb
@@ -47,6 +47,8 @@ class Configuration
                 :default_exclude_links,
                 :default_resource_retrieval_strategy,
                 :use_related_resource_records_for_joins,
+                :default_find_related_through,
+                :default_find_related_through_polymorphic,
                 :related_identities_set
 
     def initialize
@@ -181,13 +183,36 @@ def initialize
       # per resource (or base resource) using the class method `load_resource_retrieval_strategy`.
       #
       # Available strategies:
-      # 'JSONAPI::ActiveRelationRetrieval'
-      # 'JSONAPI::ActiveRelationRetrievalV09'
-      # 'JSONAPI::ActiveRelationRetrievalV10'
+      # 'JSONAPI::ActiveRelationRetrieval' - This is the default strategy. In addition, this strategy allows for will
+      # use a single phased approach to retrieve primary resources when caching is not enabled for a resource class.
+      # When caching is enabled, the retrieval of the primary resources is a two phased approach. The first phase gets
+      # the ids and cache fields. The second phase gets any cache misses from the related resource. Retrieval of related
+      # resources is configurable with the `default_find_related_through` and `default_find_related_through_polymorphic`
+      # described below.
+      # 'JSONAPI::ActiveRelationRetrievalV09' - Retrieves resources using the v0.9.x approach. This uses rails'
+      # `includes` method to retrieve related models. This requires overriding the `records_for` method on the resource
+      # to control filtering of included resources.
+      # 'JSONAPI::ActiveRelationRetrievalV10' - Retrieves resources using the v0.10.x approach. This always retrieves
+      # related resources through the primary resource joined to the related resource (through_primary).
+      # Custom - Specify the a custom retrieval strategy module name as a string
       # :none
       # :self
       self.default_resource_retrieval_strategy = 'JSONAPI::ActiveRelationRetrieval'
 
+      # For 'JSONAPI::ActiveRelationRetrieval' we can refine how related resources are retrieved with options for
+      # monomorphic and polymorphic relationships. The default is :inverse for both.
+      # :inverse - use the inverse relationship on the related resource. This joins the related resource to the
+      # primary resource table. To use this a relationship to the primary resource must be defined on the related
+      # resource.
+      # :primary - use the primary resource joined with the related resources table. This results in a two phased
+      # querying approach. The first phase gets the ids and cache fields. The second phase gets any cache misses
+      # from the related resource. In the second phase permissions are not applied since they were already applied in
+      # the first phase. This behavior is consistent with JR v0.10.x, with the exception that when caching is disabled
+      # the retrieval of the primary resources does not need to be done in two phases.
+
+      self.default_find_related_through = :inverse
+      self.default_find_related_through_polymorphic = :inverse
+
       # For 'JSONAPI::ActiveRelationRetrievalV10': use a related resource's `records` when performing joins.
       # This setting allows included resources to account for permission scopes. It can be overridden explicitly per
       # relationship. Furthermore, specifying a `relation_name` on a relationship will cause this setting to be ignored.
@@ -359,6 +384,10 @@ def allow_include=(allow_include)
 
     attr_writer :use_related_resource_records_for_joins
 
+    attr_writer :default_find_related_through
+
+    attr_writer :default_find_related_through_polymorphic
+
     attr_writer :related_identities_set
   end
 
diff --git a/lib/jsonapi/processor.rb b/lib/jsonapi/processor.rb
index 6d46f000..b07dc11e 100644
--- a/lib/jsonapi/processor.rb
+++ b/lib/jsonapi/processor.rb
@@ -292,7 +292,7 @@ def replace_to_one_relationship
       key_value = params[:key_value]
 
       resource = resource_klass.find_by_key(resource_id, context: context)
-      result = resource.replace_to_one_link(relationship_type, key_value)
+      result = resource.replace_to_one_link(relationship_type, key_value, context: context)
 
       JSONAPI::OperationResult.new(result == :completed ? :no_content : :accepted, result_options)
     end
@@ -304,7 +304,7 @@ def replace_polymorphic_to_one_relationship
       key_type = params[:key_type]
 
       resource = resource_klass.find_by_key(resource_id, context: context)
-      result = resource.replace_polymorphic_to_one_link(relationship_type, key_value, key_type)
+      result = resource.replace_polymorphic_to_one_link(relationship_type, key_value, key_type, context: context)
 
       JSONAPI::OperationResult.new(result == :completed ? :no_content : :accepted, result_options)
     end
@@ -315,7 +315,7 @@ def create_to_many_relationships
       data = params[:data]
 
       resource = resource_klass.find_by_key(resource_id, context: context)
-      result = resource.create_to_many_links(relationship_type, data)
+      result = resource.create_to_many_links(relationship_type, data, context: context)
 
       JSONAPI::OperationResult.new(result == :completed ? :no_content : :accepted, result_options)
     end
@@ -326,7 +326,7 @@ def replace_to_many_relationships
       data = params.fetch(:data)
 
       resource = resource_klass.find_by_key(resource_id, context: context)
-      result = resource.replace_to_many_links(relationship_type, data)
+      result = resource.replace_to_many_links(relationship_type, data, context: context)
 
       JSONAPI::OperationResult.new(result == :completed ? :no_content : :accepted, result_options)
     end
@@ -353,7 +353,7 @@ def remove_to_one_relationship
       relationship_type = params[:relationship_type].to_sym
 
       resource = resource_klass.find_by_key(resource_id, context: context)
-      result = resource.remove_to_one_link(relationship_type)
+      result = resource.remove_to_one_link(relationship_type, context: context)
 
       JSONAPI::OperationResult.new(result == :completed ? :no_content : :accepted, result_options)
     end
diff --git a/lib/jsonapi/relationship.rb b/lib/jsonapi/relationship.rb
index 8f7d0fb1..ea12231b 100644
--- a/lib/jsonapi/relationship.rb
+++ b/lib/jsonapi/relationship.rb
@@ -2,14 +2,25 @@
 
 module JSONAPI
   class Relationship
-    attr_reader :acts_as_set, :foreign_key, :options, :name,
-                :class_name, :polymorphic, :always_include_optional_linkage_data, :exclude_linkage_data,
-                :parent_resource, :eager_load_on_include, :custom_methods,
-                :inverse_relationship, :allow_include, :hidden, :use_related_resource_records_for_joins
-
-    attr_writer :allow_include
-
-    attr_accessor :_routed, :_warned_missing_route, :_warned_missing_inverse_relationship
+    attr_reader :acts_as_set,
+                :foreign_key,
+                :options,
+                :name,
+                :class_name,
+                :polymorphic,
+                :always_include_optional_linkage_data,
+                :exclude_linkage_data,
+                :parent_resource,
+                :eager_load_on_include,
+                :custom_methods,
+                :inverse_relationship,
+                :hidden,
+                :use_related_resource_records_for_joins,
+                :find_related_through
+
+    attr_accessor :allow_include,
+                  :_routed,
+                  :_warned_missing_route, :_warned_missing_inverse_relationship
 
     def initialize(name, options = {})
       @name = name.to_s
@@ -43,6 +54,9 @@ def initialize(name, options = {})
       @allow_include = options[:allow_include]
       @class_name = nil
 
+      find_related_through = options.fetch(:find_related_through, parent_resource_klass&.default_find_related_through)
+      @find_related_through = find_related_through&.to_sym
+
       @inverse_relationship = options[:inverse_relationship]&.to_sym
 
       @_routed = false
@@ -88,6 +102,10 @@ def inverse_relationship
       @inverse_relationship
     end
 
+    def inverse_relationship_klass
+      @inverse_relationship_klass ||= resource_klass._relationship(inverse_relationship)
+    end
+
     def polymorphic_types
       return @polymorphic_types if @polymorphic_types
 
diff --git a/lib/jsonapi/resource_common.rb b/lib/jsonapi/resource_common.rb
index 9ec4e0d1..7f06fc1c 100644
--- a/lib/jsonapi/resource_common.rb
+++ b/lib/jsonapi/resource_common.rb
@@ -425,11 +425,15 @@ def _replace_fields(field_data)
       :completed
     end
 
-    def find_related_ids(relationship, options = {})
+    def find_related_ids(relationship, _options)
       send(relationship.foreign_key)
     end
 
     module ClassMethods
+      def default_find_related_through(_polymorphic = false)
+        nil
+      end
+
       def resource_retrieval_strategy(module_name = JSONAPI.configuration.default_resource_retrieval_strategy)
         module_name = module_name.to_s
 
@@ -710,11 +714,11 @@ def has_many(*attrs)
       # ```
       # so in order to invoke the right class from subclasses,
       # we should call this method to override it.
-      def model_name(model, options = {})
+      def model_name(model, add_model_hint: true)
         @model_class = nil
         @_model_name = model.to_sym
 
-        model_hint(model: @_model_name, resource: self) unless options[:add_model_hint] == false
+        model_hint(model: @_model_name, resource: self) if add_model_hint
 
         rebuild_relationships(_relationships)
       end
diff --git a/lib/jsonapi/resource_set.rb b/lib/jsonapi/resource_set.rb
index e5846994..b28ac5e6 100644
--- a/lib/jsonapi/resource_set.rb
+++ b/lib/jsonapi/resource_set.rb
@@ -39,12 +39,12 @@ def populate!(serializer, context, options)
       @resource_klasses.each_key do |resource_klass|
         missed_resource_ids[resource_klass] ||= []
 
-        serializer_config_key = serializer.config_key(resource_klass).gsub("/", "_")
-        context_json = resource_klass.attribute_caching_context(context).to_json
-        context_b64 = JSONAPI.configuration.resource_cache_digest_function.call(context_json)
-        context_key = "ATTR-CTX-#{context_b64.gsub("/", "_")}"
-
         if resource_klass.caching?
+          serializer_config_key = serializer.config_key(resource_klass).gsub("/", "_")
+          context_json = resource_klass.attribute_caching_context(context).to_json
+          context_b64 = JSONAPI.configuration.resource_cache_digest_function.call(context_json)
+          context_key = "ATTR-CTX-#{context_b64.gsub("/", "_")}"
+
           cache_ids = @resource_klasses[resource_klass].map do |(k, v)|
             # Store the hashcode of the cache_field to avoid storing objects and to ensure precision isn't lost
             # on timestamp types (i.e. string conversions dropping milliseconds)
diff --git a/test/controllers/controller_test.rb b/test/controllers/controller_test.rb
index 2fe181d4..ef555425 100644
--- a/test/controllers/controller_test.rb
+++ b/test/controllers/controller_test.rb
@@ -296,7 +296,18 @@ def test_index_filter_not_allowed
   end
 
   def test_index_include_one_level_query_count
-    assert_query_count(testing_v10? ? 4 : 2) do
+    expected_count = case
+                     when testing_v09?
+                       2
+                     when testing_v10?
+                       4
+                     when through_primary?
+                       3
+                     else
+                       2
+                     end
+
+    assert_query_count(expected_count) do
       assert_cacheable_get :index, params: {include: 'author'}
     end
 
@@ -304,7 +315,18 @@ def test_index_include_one_level_query_count
   end
 
   def test_index_include_two_levels_query_count
-    assert_query_count(testing_v10? ? 6 : 3) do
+    expected_count = case
+                     when testing_v09?
+                       3
+                     when testing_v10?
+                       6
+                     when through_primary?
+                       5
+                     else
+                       3
+                     end
+
+    assert_query_count(expected_count) do
       assert_cacheable_get :index, params: { include: 'author,author.comments' }
     end
     assert_response :success
@@ -3303,7 +3325,18 @@ def test_books_offset_pagination_no_params_includes_query_count_one_level
 
     with_jsonapi_config_changes do
       JSONAPI.configuration.json_key_format = :dasherized_key
-      assert_query_count(testing_v10? ? 5 : 3) do
+      expected_count = case
+                       when testing_v09?
+                         3
+                       when testing_v10?
+                         5
+                       when through_primary?
+                         4
+                       else
+                         3
+                       end
+
+      assert_query_count(expected_count) do
         assert_cacheable_get :index, params: { include: 'book-comments' }
       end
       assert_response :success
@@ -3317,7 +3350,18 @@ def test_books_offset_pagination_no_params_includes_query_count_two_levels
     with_jsonapi_config_changes do
       JSONAPI.configuration.json_key_format = :dasherized_key
 
-      assert_query_count(testing_v10? ? 7 : 4) do
+      expected_count = case
+                       when testing_v09?
+                         4
+                       when testing_v10?
+                         7
+                       when through_primary?
+                         6
+                       else
+                         4
+                       end
+
+      assert_query_count(expected_count) do
         assert_cacheable_get :index, params: { include: 'book-comments,book-comments.author' }
       end
       assert_response :success
@@ -3451,7 +3495,18 @@ def test_books_included_paged
 
     with_jsonapi_config_changes do
       JSONAPI.configuration.json_key_format = :dasherized_key
-      assert_query_count(testing_v10? ? 5 : 3) do
+      expected_count = case
+                       when testing_v09?
+                         3
+                       when testing_v10?
+                         5
+                       when through_primary?
+                         4
+                       else
+                         3
+                       end
+
+      assert_query_count(expected_count) do
         assert_cacheable_get :index, params: { filter: { id: '0' }, include: 'book-comments' }
         assert_response :success
         assert_equal 1, json_response['data'].size
@@ -3485,7 +3540,19 @@ def test_books_banned_non_book_admin_includes_switched
 
       Api::V2::BookResource.paginator :offset
       JSONAPI.configuration.top_level_meta_include_record_count = true
-      assert_query_count(testing_v10? ? 5 : 3) do
+
+      expected_count = case
+                       when testing_v09?
+                         3
+                       when testing_v10?
+                         5
+                       when through_primary?
+                         4
+                       else
+                         3
+                       end
+
+      assert_query_count(expected_count) do
         assert_cacheable_get :index, params: { page: { offset: 0, limit: 12 }, include: 'book-comments' }
         assert_response :success
         assert_equal 12, json_response['data'].size
@@ -3504,7 +3571,19 @@ def test_books_banned_non_book_admin_includes_nested_includes
       JSONAPI.configuration.json_key_format = :dasherized_key
       JSONAPI.configuration.top_level_meta_include_record_count = true
       Api::V2::BookResource.paginator :offset
-      assert_query_count(testing_v10? ? 7 : 4) do
+
+      expected_count = case
+                       when testing_v09?
+                         4
+                       when testing_v10?
+                         7
+                       when through_primary?
+                         6
+                       else
+                         4
+                       end
+
+      assert_query_count(expected_count) do
         assert_cacheable_get :index, params: { page: { offset: 0, limit: 12 }, include: 'book-comments.author' }
         assert_response :success
         assert_equal 12, json_response['data'].size
@@ -3574,7 +3653,18 @@ def test_books_included_exclude_unapproved
     with_jsonapi_config_changes do
       JSONAPI.configuration.json_key_format = :dasherized_key
 
-      assert_query_count(testing_v10? ? 4 : 2) do
+      expected_count = case
+                       when testing_v09?
+                         2
+                       when testing_v10?
+                         4
+                       when through_primary?
+                         3
+                       else
+                         2
+                       end
+
+      assert_query_count(expected_count) do
         assert_cacheable_get :index, params: { filter: { id: '0,1,2,3,4' }, include: 'book-comments' }
       end
       assert_response :success
diff --git a/test/fixtures/active_record.rb b/test/fixtures/active_record.rb
index cdcd8a65..7b80c11d 100644
--- a/test/fixtures/active_record.rb
+++ b/test/fixtures/active_record.rb
@@ -1333,7 +1333,7 @@ class SpecialBaseResource < BaseResource
 class SpecialPersonResource < SpecialBaseResource
   model_name 'Person'
 
-  def self.records(options = {})
+  def self.records(options)
     Person.where(special: true)
   end
 end
@@ -1551,19 +1551,19 @@ def find_records(filters, options)
       EOF
     end
 
-    def find_record_by_key(key, options = {})
+    def find_record_by_key(key, options)
       fail NotImplementedError, <<~EOF
         Should be something like
-        def find_record_by_key(key, options = {})
+        def find_record_by_key(key, options)
           $breed_data.breeds[key.to_i]
         end
       EOF
     end
 
-    def find_records_by_keys(keys, options = {})
+    def find_records_by_keys(keys, options)
       fail NotImplementedError, <<~EOF
         Should be something like
-        def find_records_by_keys(keys, options = {})
+        def find_records_by_keys(keys, options)
           breeds = []
           keys.each do |key|
             breeds.push($breed_data.breeds[key.to_i])
@@ -1581,7 +1581,7 @@ def find_records_by_keys(keys, options = {})
     # @option options [Hash] :include_directives The `include_directives`
     #
     # @return [Array<Resource>] the Resource instances matching the filters, sorting and pagination rules.
-    def find(filters, options = {})
+    def find(filters, options)
       records = find_records(filters, options)
       resources_for(records, options[:context])
     end
@@ -1611,7 +1611,7 @@ def resource_klass
     # @option options [Hash] :context The context of the request, set in the controller
     #
     # @return [Integer] the count
-    def count(filters, options = {})
+    def count(filters, options)
       fail NotImplementedError, <<~EOF
         Should be something like
         def count(filters, options)
@@ -1624,12 +1624,12 @@ def count(filters, options)
     #
     # @param key the primary key of the resource to find
     # @option options [Hash] :context The context of the request, set in the controller
-    def find_by_key(key, options = {})
+    def find_by_key(key, options)
       record = find_record_by_key(key, options)
       resource_for(record, options[:context])
     end
 
-    def find_to_populate_by_keys(keys, options = {})
+    def find_to_populate_by_keys(keys, options)
       find_by_keys(keys, options)
     end
 
@@ -1637,7 +1637,7 @@ def find_to_populate_by_keys(keys, options = {})
     #
     # @param keys [Array<key>] Array of primary keys to find resources for
     # @option options [Hash] :context The context of the request, set in the controller
-    def find_by_keys(keys, options = {})
+    def find_by_keys(keys, options)
       records = find_records_by_keys(keys, options)
       resources_for(records, options[:context])
     end
@@ -1651,7 +1651,7 @@ class BreedResource < PoroResource
   routing_options param: :id
 
   class << self
-    def find_records(filters, options = {})
+    def find_records(filters, options)
       breeds = []
       id_filter = filters[:id]
       id_filter = [id_filter] unless id_filter.nil? || id_filter.is_a?(Array)
@@ -1661,11 +1661,11 @@ def find_records(filters, options = {})
       breeds
     end
 
-    def find_record_by_key(key, options = {})
+    def find_record_by_key(key, options)
       $breed_data.breeds[key.to_i]
     end
 
-    def find_records_by_keys(keys, options = {})
+    def find_records_by_keys(keys, options)
       breeds = []
       keys.each do |key|
         breeds.push($breed_data.breeds[key.to_i])
@@ -1959,7 +1959,7 @@ class BookResource < JSONAPI::Resource
 
       has_many "authors", class_name: 'Authors'
 
-      has_many "book_comments", relation_name: -> (options = {}) {
+      has_many "book_comments", relation_name: -> (options) {
         context = options[:context]
         current_user = context ? context[:current_user] : nil
 
@@ -1999,7 +1999,7 @@ def not_banned_books
           books[:banned].eq(false)
         end
 
-        def records(options = {})
+        def records(options)
           context = options[:context]
           current_user = context ? context[:current_user] : nil
 
@@ -2052,7 +2052,7 @@ def approved_comments(approved = true)
           book_comments[:approved].eq(approved)
         end
 
-        def records(options = {})
+        def records(options)
           current_user = options[:context][:current_user]
           _model_class.for_user(current_user)
         end
@@ -2081,7 +2081,7 @@ module Api
   module V4
     class PostResource < PostResource
       class << self
-        def records(options = {})
+        def records(options)
           # Sets up a performance issue for testing
           super(options).joins(:comments)
         end
diff --git a/test/test_helper.rb b/test/test_helper.rb
index c61f8d2e..d0d44699 100644
--- a/test/test_helper.rb
+++ b/test/test_helper.rb
@@ -699,6 +699,10 @@ def assert_cacheable_get(action, **args)
     @queries = orig_queries
   end
 
+  def through_primary?
+    JSONAPI.configuration.default_find_related_through == :primary
+  end
+
   def testing_v10?
     JSONAPI.configuration.default_resource_retrieval_strategy == 'JSONAPI::ActiveRelationRetrievalV10'
   end
diff --git a/test/unit/active_relation_resource_finder/join_manager_test.rb b/test/unit/active_relation_resource_finder/join_manager_through_inverse_test.rb
similarity index 87%
rename from test/unit/active_relation_resource_finder/join_manager_test.rb
rename to test/unit/active_relation_resource_finder/join_manager_through_inverse_test.rb
index 53799e9e..802b052b 100644
--- a/test/unit/active_relation_resource_finder/join_manager_test.rb
+++ b/test/unit/active_relation_resource_finder/join_manager_through_inverse_test.rb
@@ -1,7 +1,7 @@
 require File.expand_path('../../../test_helper', __FILE__)
 require 'jsonapi-resources'
 
-class JoinManagerTest < ActiveSupport::TestCase
+class JoinManagerThroughInverseTest < ActiveSupport::TestCase
   # def setup
   #   JSONAPI.configuration.default_alias_on_join = false
   # end
@@ -11,7 +11,7 @@ class JoinManagerTest < ActiveSupport::TestCase
   # end
 
   def test_no_added_joins
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PostResource)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PostResource)
 
     records = PostResource.records({})
     records = join_manager.join(records, {})
@@ -22,7 +22,7 @@ def test_no_added_joins
 
   def test_add_single_join
     filters = {'tags' => ['1']}
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PostResource, filters: filters)
     records = PostResource.records({})
     records = join_manager.join(records, {})
     assert_equal 'SELECT "posts".* FROM "posts" LEFT OUTER JOIN "posts_tags" ON "posts_tags"."post_id" = "posts"."id" LEFT OUTER JOIN "tags" ON "tags"."id" = "posts_tags"."tag_id"', sql_for_compare(records.to_sql)
@@ -32,7 +32,7 @@ def test_add_single_join
 
   def test_add_single_sort_join
     sort_criteria = [{field: 'tags.name', direction: :desc}]
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PostResource, sort_criteria: sort_criteria)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PostResource, sort_criteria: sort_criteria)
     records = PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -44,7 +44,7 @@ def test_add_single_sort_join
   def test_add_single_sort_and_filter_join
     filters = {'tags' => ['1']}
     sort_criteria = [{field: 'tags.name', direction: :desc}]
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PostResource, sort_criteria: sort_criteria, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PostResource, sort_criteria: sort_criteria, filters: filters)
     records = PostResource.records({})
     records = join_manager.join(records, {})
     assert_equal 'SELECT "posts".* FROM "posts" LEFT OUTER JOIN "posts_tags" ON "posts_tags"."post_id" = "posts"."id" LEFT OUTER JOIN "tags" ON "tags"."id" = "posts_tags"."tag_id"', sql_for_compare(records.to_sql)
@@ -58,7 +58,7 @@ def test_add_sibling_joins
         'author' => ['1']
     }
 
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PostResource, filters: filters)
     records = PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -70,7 +70,7 @@ def test_add_sibling_joins
 
 
   def test_add_joins_source_relationship
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PostResource,
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PostResource,
                                                                           source_relationship: PostResource._relationship(:comments))
     records = PostResource.records({})
     records = join_manager.join(records, {})
@@ -81,7 +81,7 @@ def test_add_joins_source_relationship
 
 
   def test_add_joins_source_relationship_with_custom_apply
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: Api::V10::PostResource,
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: Api::V10::PostResource,
                                                                           source_relationship: Api::V10::PostResource._relationship(:comments))
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
@@ -100,7 +100,7 @@ def test_add_nested_scoped_joins
         'author' => ['1']
     }
 
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: Api::V10::PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: Api::V10::PostResource, filters: filters)
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -117,7 +117,7 @@ def test_add_nested_scoped_joins
         'comments.tags' => ['1']
     }
 
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: Api::V10::PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: Api::V10::PostResource, filters: filters)
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -135,7 +135,7 @@ def test_add_nested_joins_with_fields
         'author.foo' => ['1']
     }
 
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: Api::V10::PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: Api::V10::PostResource, filters: filters)
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -149,7 +149,7 @@ def test_add_nested_joins_with_fields
   def test_add_joins_with_sub_relationship
     relationships = %w(author author.comments tags)
 
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: Api::V10::PostResource, relationships: relationships,
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: Api::V10::PostResource, relationships: relationships,
                                                                           source_relationship: Api::V10::PostResource._relationship(:comments))
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
@@ -168,7 +168,7 @@ def test_add_joins_with_sub_relationship_and_filters
 
     relationships = %w(author author.comments tags)
 
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PostResource,
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PostResource,
                                                                           filters: filters,
                                                                           relationships: relationships,
                                                                           source_relationship: PostResource._relationship(:comments))
@@ -183,7 +183,7 @@ def test_add_joins_with_sub_relationship_and_filters
   end
 
   def test_polymorphic_join_belongs_to_just_source
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(
       resource_klass: PictureResource,
       source_relationship: PictureResource._relationship(:imageable)
     )
@@ -200,7 +200,7 @@ def test_polymorphic_join_belongs_to_just_source
 
   def test_polymorphic_join_belongs_to_filter
     filters = {'imageable' => ['Foo']}
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PictureResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PictureResource, filters: filters)
 
     records = PictureResource.records({})
     records = join_manager.join(records, {})
@@ -217,7 +217,7 @@ def test_polymorphic_join_belongs_to_filter_on_resource
     }
 
     relationships = %w(imageable file_properties)
-    join_manager = JSONAPI::ActiveRelation::JoinManager.new(resource_klass: PictureResource,
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughInverse.new(resource_klass: PictureResource,
                                                                           filters: filters,
                                                                           relationships: relationships)
 
diff --git a/test/unit/active_relation_resource_finder/join_manager_v10_test.rb b/test/unit/active_relation_resource_finder/join_manager_through_primary_test.rb
similarity index 81%
rename from test/unit/active_relation_resource_finder/join_manager_v10_test.rb
rename to test/unit/active_relation_resource_finder/join_manager_through_primary_test.rb
index d0ec9a29..0122c7fb 100644
--- a/test/unit/active_relation_resource_finder/join_manager_v10_test.rb
+++ b/test/unit/active_relation_resource_finder/join_manager_through_primary_test.rb
@@ -1,9 +1,9 @@
 require File.expand_path('../../../test_helper', __FILE__)
 require 'jsonapi-resources'
 
-class JoinManagerV10Test < ActiveSupport::TestCase
+class JoinManagerThroughPrimaryTest < ActiveSupport::TestCase
   def test_no_added_joins
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PostResource)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PostResource)
 
     records = PostResource.records({})
     records = join_manager.join(records, {})
@@ -14,7 +14,7 @@ def test_no_added_joins
 
   def test_add_single_join
     filters = {'tags' => ['1']}
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PostResource, filters: filters)
     records = PostResource.records({})
     records = join_manager.join(records, {})
     assert_equal 'SELECT "posts".* FROM "posts" LEFT OUTER JOIN "posts_tags" ON "posts_tags"."post_id" = "posts"."id" LEFT OUTER JOIN "tags" ON "tags"."id" = "posts_tags"."tag_id"', sql_for_compare(records.to_sql)
@@ -24,7 +24,7 @@ def test_add_single_join
 
   def test_joins_have_join_options
     filters = {'tags' => ['1']}
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PostResource, filters: filters)
     records = PostResource.records({})
     records = join_manager.join(records, {})
     assert_equal 'SELECT "posts".* FROM "posts" LEFT OUTER JOIN "posts_tags" ON "posts_tags"."post_id" = "posts"."id" LEFT OUTER JOIN "tags" ON "tags"."id" = "posts_tags"."tag_id"', sql_for_compare(records.to_sql)
@@ -38,7 +38,7 @@ def test_joins_have_join_options
 
   def test_add_single_sort_join
     sort_criteria = [{field: 'tags.name', direction: :desc}]
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PostResource, sort_criteria: sort_criteria)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PostResource, sort_criteria: sort_criteria)
     records = PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -50,7 +50,7 @@ def test_add_single_sort_join
   def test_add_single_sort_and_filter_join
     filters = {'tags' => ['1']}
     sort_criteria = [{field: 'tags.name', direction: :desc}]
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PostResource, sort_criteria: sort_criteria, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PostResource, sort_criteria: sort_criteria, filters: filters)
     records = PostResource.records({})
     records = join_manager.join(records, {})
     assert_equal 'SELECT "posts".* FROM "posts" LEFT OUTER JOIN "posts_tags" ON "posts_tags"."post_id" = "posts"."id" LEFT OUTER JOIN "tags" ON "tags"."id" = "posts_tags"."tag_id"', sql_for_compare(records.to_sql)
@@ -64,7 +64,7 @@ def test_add_sibling_joins
       'author' => ['1']
     }
 
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PostResource, filters: filters)
     records = PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -76,8 +76,8 @@ def test_add_sibling_joins
 
 
   def test_add_joins_source_relationship
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PostResource,
-                                                            source_relationship: PostResource._relationship(:comments))
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PostResource,
+                                                                          source_relationship: PostResource._relationship(:comments))
     records = PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -87,8 +87,8 @@ def test_add_joins_source_relationship
 
 
   def test_add_joins_source_relationship_with_custom_apply
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: Api::V10::PostResource,
-                                                            source_relationship: Api::V10::PostResource._relationship(:comments))
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: Api::V10::PostResource,
+                                                                          source_relationship: Api::V10::PostResource._relationship(:comments))
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -106,7 +106,7 @@ def test_add_nested_scoped_joins
       'author' => ['1']
     }
 
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: Api::V10::PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: Api::V10::PostResource, filters: filters)
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -123,7 +123,7 @@ def test_add_nested_scoped_joins
       'comments.tags' => ['1']
     }
 
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: Api::V10::PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: Api::V10::PostResource, filters: filters)
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -141,7 +141,7 @@ def test_add_nested_joins_with_fields
       'author.foo' => ['1']
     }
 
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: Api::V10::PostResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: Api::V10::PostResource, filters: filters)
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -155,8 +155,8 @@ def test_add_nested_joins_with_fields
   def test_add_joins_with_sub_relationship
     relationships = %w(author author.comments tags)
 
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: Api::V10::PostResource, relationships: relationships,
-                                                            source_relationship: Api::V10::PostResource._relationship(:comments))
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: Api::V10::PostResource, relationships: relationships,
+                                                                          source_relationship: Api::V10::PostResource._relationship(:comments))
     records = Api::V10::PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -174,10 +174,10 @@ def test_add_joins_with_sub_relationship_and_filters
 
     relationships = %w(author author.comments tags)
 
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PostResource,
-                                                            filters: filters,
-                                                            relationships: relationships,
-                                                            source_relationship: PostResource._relationship(:comments))
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PostResource,
+                                                                          filters: filters,
+                                                                          relationships: relationships,
+                                                                          source_relationship: PostResource._relationship(:comments))
     records = PostResource.records({})
     records = join_manager.join(records, {})
 
@@ -189,8 +189,8 @@ def test_add_joins_with_sub_relationship_and_filters
   end
 
   def test_polymorphic_join_belongs_to_just_source
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PictureResource,
-                                                            source_relationship: PictureResource._relationship(:imageable))
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PictureResource,
+                                                                          source_relationship: PictureResource._relationship(:imageable))
 
     records = PictureResource.records({})
     records = join_manager.join(records, {})
@@ -204,7 +204,7 @@ def test_polymorphic_join_belongs_to_just_source
 
   def test_polymorphic_join_belongs_to_filter
     filters = {'imageable' => ['Foo']}
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PictureResource, filters: filters)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PictureResource, filters: filters)
 
     records = PictureResource.records({})
     records = join_manager.join(records, {})
@@ -221,9 +221,9 @@ def test_polymorphic_join_belongs_to_filter_on_resource
     }
 
     relationships = %w(imageable file_properties)
-    join_manager = JSONAPI::ActiveRelation::JoinManagerV10.new(resource_klass: PictureResource,
-                                                            filters: filters,
-                                                            relationships: relationships)
+    join_manager = JSONAPI::ActiveRelation::JoinManagerThroughPrimary.new(resource_klass: PictureResource,
+                                                                          filters: filters,
+                                                                          relationships: relationships)
 
     records = PictureResource.records({})
     records = join_manager.join(records, {})
diff --git a/test/unit/resource/active_relation_resource_v_11_test.rb b/test/unit/resource/active_relation_resource_v_11_test.rb
index f9ae49b6..b55f7895 100644
--- a/test/unit/resource/active_relation_resource_v_11_test.rb
+++ b/test/unit/resource/active_relation_resource_v_11_test.rb
@@ -68,7 +68,8 @@ def setup
 
   def test_find_fragments_no_attributes
     filters = {}
-    posts_identities = V11::PostResource.find_fragments(filters)
+    options = {}
+    posts_identities = V11::PostResource.find_fragments(filters, options)
 
     assert_equal 20, posts_identities.length
     assert_equal JSONAPI::ResourceIdentity.new(V11::PostResource, 1), posts_identities.keys[0]
diff --git a/test/unit/serializer/serializer_test.rb b/test/unit/serializer/serializer_test.rb
index 47e5518a..072a6385 100644
--- a/test/unit/serializer/serializer_test.rb
+++ b/test/unit/serializer/serializer_test.rb
@@ -905,7 +905,8 @@ def test_serializer_include_from_resource
 
       directives = JSONAPI::IncludeDirectives.new(PostResource, ['author'])
 
-      resource_set = JSONAPI::ResourceSet.new(PostResource.find_by_key(1), directives[:include_related], {})
+      options = {}
+      resource_set = JSONAPI::ResourceSet.new(PostResource.find_by_key(1, options), directives[:include_related], {})
       resource_set.populate!(serializer, {}, {})
 
       serialized = serializer.serialize_resource_set_to_hash_single(resource_set)