diff --git a/lib/mongoid/association/eager.rb b/lib/mongoid/association/eager.rb
index 4ddea6622e..5e32477229 100644
--- a/lib/mongoid/association/eager.rb
+++ b/lib/mongoid/association/eager.rb
@@ -4,6 +4,15 @@ module Mongoid
   module Association
     # Base class for eager load preload functions.
     class Eager
+      # Build a preloader for the given arguments and run it.
+      #
+      # @param (see #initialize)
+      #
+      # @return [ Array ] The list of documents given.
+      def self.run(associations, docs, use_lookup = false, pipeline = [])
+        new(associations, docs, use_lookup, pipeline).run
+      end
+
       # Instantiate the eager load class.
       #
       # @example Create the new belongs to eager load preloader.
diff --git a/lib/mongoid/association/eager_load/discriminated_inclusion.rb b/lib/mongoid/association/eager_load/discriminated_inclusion.rb
new file mode 100644
index 0000000000..106ae54653
--- /dev/null
+++ b/lib/mongoid/association/eager_load/discriminated_inclusion.rb
@@ -0,0 +1,88 @@
+# frozen_string_literal: true
+
+require 'mongoid/association/eager_load/inclusion'
+
+module Mongoid
+  module Association
+    module EagerLoad
+      # Loads an inclusion that more than one subclass defines under the same name
+      # but pointing at different targets. A single $lookup can't serve them: they
+      # would all write to the same field and overwrite one another. So each
+      # subclass's inclusion is contributed into its own temporary field, carrying
+      # its own nested children, and a $set then routes every document to the field
+      # for its own type, by the discriminator.
+      #
+      # For Machine.eager_load(:widgets), where Lathe#widgets => Cog and
+      # Press#widgets => Belt, it emits:
+      #
+      #   { '$lookup' => { 'from' => 'cogs',  ..., 'as' => '__eager_load_widgets_Lathe' } },
+      #   { '$lookup' => { 'from' => 'belts', ..., 'as' => '__eager_load_widgets_Press' } },
+      #   { '$set' => {
+      #     'widgets' => { '$switch' => { 'branches' => [   # route each document to its
+      #       { 'case' => { '$eq' => [ '$_type', 'Lathe' ] }, 'then' => '$__eager_load_widgets_Lathe' },  # own type's matches
+      #       { 'case' => { '$eq' => [ '$_type', 'Press' ] }, 'then' => '$__eager_load_widgets_Press' }
+      #     ], 'default' => [] } }
+      #   } },
+      #   { '$unset' => [ '__eager_load_widgets_Lathe', '__eager_load_widgets_Press' ] }
+      #
+      # @api private
+      class DiscriminatedInclusion < Inclusion
+        def initialize(nodes)
+          super()
+          @nodes = nodes
+        end
+
+        # Append each subclass's lookup (into its own temporary field), the routing
+        # $set, and the cleanup $unset.
+        #
+        # @param [ Array<Hash> ] destination The pipeline the stages are appended to.
+        def contribute(destination, _chain)
+          fields = @nodes.map { |node| [ node, contribute_into_temporary(destination, node) ] }
+          destination << route_by_type(fields)
+          destination << { '$unset' => fields.map { |_node, field| field } }
+        end
+
+        private
+
+        # Let the node build its own $lookup (with its nested children) and redirect
+        # it to write into a temporary field instead of the shared association name.
+        def contribute_into_temporary(destination, node)
+          field = temporary_field(node)
+          captured = []
+          node.contribute(captured, [])
+          captured.first['$lookup']['as'] = field
+          destination.concat(captured)
+          field
+        end
+
+        def temporary_field(node)
+          "__eager_load_#{node.association.name}_#{owner(node).discriminator_value}"
+        end
+
+        # The $set that fills the association on each document from the temporary
+        # field matching its own type.
+        def route_by_type(fields)
+          branches = fields.map do |node, field|
+            {
+              'case' => { '$eq' => [ "$#{discriminator_key}", owner(node).discriminator_value ] },
+              'then' => "$#{field}"
+            }
+          end
+          { '$set' => { name => { '$switch' => { 'branches' => branches, 'default' => [] } } } }
+        end
+
+        def owner(node)
+          node.association.inverse_class
+        end
+
+        def name
+          @nodes.first.association.name.to_s
+        end
+
+        def discriminator_key
+          owner(@nodes.first).discriminator_key
+        end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/association/eager_load/embedded_distributor.rb b/lib/mongoid/association/eager_load/embedded_distributor.rb
new file mode 100644
index 0000000000..ad4b6a784d
--- /dev/null
+++ b/lib/mongoid/association/eager_load/embedded_distributor.rb
@@ -0,0 +1,151 @@
+# frozen_string_literal: true
+
+module Mongoid
+  module Association
+    # Objects that carry out the aggregation-based eager load triggered by
+    # Criteria#eager_load.
+    module EagerLoad
+      # Distributes the results of a $lookup onto the embedded documents they
+      # belong to.
+      #
+      # A $lookup overwrites the single field it writes to, and it can't distribute
+      # its matches across the elements of an embedded array. So when the reference
+      # being eager-loaded lives on an embedded document, the matches are first
+      # collected in a temporary top-level field and then distributed down the
+      # embedded path onto each embedded document, merging into it (so the rest of
+      # the document is kept) and correlating by key. The temporary field is then
+      # dropped.
+      #
+      # For Computer.eager_load(port: :device) (Port belongs_to :device) it emits:
+      #
+      #   { '$lookup' => {                     # devices can't be written into
+      #     'from' => 'devices',               # the embedded port, so they are
+      #     'localField' => 'port.device_id',  # collected in a temp top-level
+      #     'foreignField' => '_id',           # field instead
+      #     'as' => '__eager_load_port_device'
+      #   } },
+      #   { '$set' => {
+      #     'port' => { '$mergeObjects' => [   # merge a 'device' key onto the port
+      #       '$port',
+      #       { 'device' => { '$filter' => { ... } } }   # this port's matches
+      #     ] }
+      #   } },
+      #   { '$unset' => '__eager_load_port_device' }  # drop the temp field
+      #
+      # @api private
+      class EmbeddedDistributor
+        # @param [ Mongoid::Association::Relatable ] association The referenced
+        #   inclusion being eager-loaded from within an embedded document.
+        # @param [ Array<Mongoid::Association::Relatable> ] chain The embedded
+        #   ancestors, from the root document inward, down to the association's owner.
+        # @param [ Hash ] lookup_stage The $lookup stage built for the association.
+        #
+        # @return [ EmbeddedDistributor ] The distributor.
+        class << self
+          def for(association:, chain:, lookup_stage:)
+            lookup = lookup_stage['$lookup']
+            new(association, chain, lookup_stage, lookup['localField'], lookup['foreignField'])
+          end
+
+          private :new
+        end
+
+        def initialize(association, chain, lookup_stage, local_field, foreign_field)
+          @association = association
+          @chain = chain
+          @lookup_stage = lookup_stage
+          @local_field = local_field
+          @foreign_field = foreign_field
+        end
+
+        # The stages that run the $lookup into a temporary field and then
+        # distribute its matches onto the embedded documents along the path.
+        #
+        # @return [ Array<Hash> ] The stages to append to the pipeline.
+        def stages
+          redirect_lookup_to_temporary_field
+          [
+            @lookup_stage,
+            { '$set' => {
+              root => distributed_value(@chain, "$#{root}")
+            } },
+            { '$unset' => temporary_field }
+          ]
+        end
+
+        private
+
+        # The $lookup runs at the top level, so it reads the local field by its
+        # full embedded path and writes the matches into the temporary field.
+        def redirect_lookup_to_temporary_field
+          lookup = @lookup_stage['$lookup']
+          lookup['localField'] = "#{path}.#{@local_field}"
+          lookup['as'] = temporary_field
+        end
+
+        def path
+          @chain.map(&:store_as).join('.')
+        end
+
+        def root
+          @chain.first.store_as
+        end
+
+        def temporary_field
+          "__eager_load_#{path.tr('.', '_')}_#{@association.name}"
+        end
+
+        # An embedded collection (embeds_many) is rebuilt with $map so each element
+        # keeps its own matches instead of collapsing onto the first; a single
+        # embedded document (embeds_one) receives its matches in place.
+        def distributed_value(chain, node)
+          head, *rest = chain
+          many = head.many?
+          element = many ? "$$#{head.store_as}" : node
+          child =
+            if rest.empty?
+              { @association.name.to_s => correlated_matches(element) }
+            else
+              segment = rest.first.store_as
+              { segment => distributed_value(rest, "#{element}.#{segment}") }
+            end
+          merged = { '$mergeObjects' => [ element, child ] }
+          return merge_into_present(node, merged) unless many
+
+          { '$map' => {
+            'input' => node,
+            'as' => head.store_as,
+            'in' => merged
+          } }
+        end
+
+        # Merge the matches into a single embedded document only when it exists, so
+        # an absent embeds_one stays absent instead of being synthesized from its
+        # matches alone.
+        def merge_into_present(node, merged)
+          { '$cond' => {
+            'if' => { '$ifNull' => [ node, false ] },
+            'then' => merged,
+            'else' => node
+          } }
+        end
+
+        # The matches that belong to a single embedded element.
+        def correlated_matches(element)
+          { '$filter' => {
+            'input' => "$#{temporary_field}",
+            'as' => 'match',
+            'cond' => { match_operator => [ "$$match.#{@foreign_field}", "#{element}.#{@local_field}" ] }
+          } }
+        end
+
+        # A has_and_belongs_to_many holds an array of foreign keys, so a match
+        # belongs when its key is among them ($in); every other association points
+        # at a single key ($eq).
+        def match_operator
+          @association.many_to_many? ? '$in' : '$eq'
+        end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/association/eager_load/inclusion.rb b/lib/mongoid/association/eager_load/inclusion.rb
new file mode 100644
index 0000000000..47cb03a264
--- /dev/null
+++ b/lib/mongoid/association/eager_load/inclusion.rb
@@ -0,0 +1,173 @@
+# frozen_string_literal: true
+
+module Mongoid
+  module Association
+    module EagerLoad
+      # Something an eager load contributes to the pipeline, in the role it plays
+      # while the pipeline is built. A root is asked to contribute and the whole
+      # tree follows by recursion. AssociationInclusion stands for a single
+      # association; DiscriminatedInclusion stands for a name several subclasses
+      # share.
+      #
+      # @api private
+      class Inclusion
+        # Add this inclusion's stages to the destination.
+        #
+        # @param [ Array<Hash> ] destination The pipeline (or sub-pipeline) the
+        #   stages are appended to.
+        # @param [ Array<Mongoid::Association::Relatable> ] chain The embedded path
+        #   accumulated from the ancestors above this inclusion (empty at the top).
+        def contribute(destination, chain)
+          raise NotImplementedError
+        end
+      end
+
+      # An inclusion that stands for a single association. The LookupPipeline holds
+      # the stage-building helpers the kinds lean on, and a node carries its own
+      # children, so the pipeline is built by recursion from the roots downward.
+      #
+      # @api private
+      class AssociationInclusion < Inclusion
+        class << self
+          # Builds the right kind of inclusion for the association. Each subclass
+          # decides whether it handles it (.for?); exactly one does.
+          #
+          # @param [ Mongoid::Association::Relatable ] association The inclusion.
+          # @param [ LookupPipeline ] pipeline The pipeline being built.
+          # @param [ Array<Inclusion> ] children The inclusions nested under it.
+          #
+          # @return [ AssociationInclusion ] The matching kind of inclusion.
+          def for(association, pipeline, children)
+            subclasses.find { |kind| kind.for?(association) }.new(association, pipeline, children)
+          end
+
+          # Whether this kind handles the given association.
+          #
+          # @param [ Mongoid::Association::Relatable ] association The inclusion.
+          #
+          # @return [ true | false ] Whether it handles it.
+          def for?(association)
+            raise NotImplementedError
+          end
+        end
+
+        # @return [ Mongoid::Association::Relatable ] The association this stands for.
+        attr_reader :association
+
+        def initialize(association, pipeline, children)
+          super()
+          @association = association
+          @pipeline = pipeline
+          @children = children
+        end
+      end
+
+      # A referenced inclusion: contributes a $lookup whose sub-pipeline holds its
+      # own children. When it lives inside an embedded document (a non-empty
+      # chain), the $lookup is distributed onto that embedded path instead of
+      # standing at the top level.
+      #
+      # For a has_many :albums it contributes:
+      #
+      #   { '$lookup' => {
+      #     'from' => 'albums',
+      #     'localField' => '_id',        # the band's _id...
+      #     'foreignField' => 'band_id',  # ...matched against each album's band_id
+      #     'as' => 'albums',             # matches are written to this field
+      #     'pipeline' => [
+      #       { '$sort' => {
+      #         '_id' => 1
+      #       } },
+      #       <children>
+      #     ]
+      #   } }
+      #
+      # @api private
+      class JoinedInclusion < AssociationInclusion
+        class << self
+          # The default kind: a referenced, non-polymorphic association, i.e. the
+          # one no sibling kind claims.
+          #
+          # @param [ Mongoid::Association::Relatable ] association The inclusion.
+          #
+          # @return [ true | false ] Whether it handles it.
+          def for?(association)
+            (superclass.subclasses - [ self ]).none? { |kind| kind.for?(association) }
+          end
+        end
+
+        # Append the $lookup, with the children in its sub-pipeline, to the
+        # destination; or distribute it onto the embedded path when nested in one.
+        #
+        # @param [ Array<Hash> ] destination The pipeline (or sub-pipeline) the
+        #   stages are appended to.
+        # @param [ Array<Mongoid::Association::Relatable> ] chain The embedded path
+        #   accumulated from the ancestors above this inclusion (empty at the top).
+        def contribute(destination, chain)
+          stage = @pipeline.lookup_stage_for(@association)
+          @children.each { |child| child.contribute(stage['$lookup']['pipeline'], []) }
+
+          if chain.empty?
+            destination << stage
+          else
+            destination.concat(@pipeline.distribute(@association, chain, stage))
+          end
+        end
+      end
+
+      # An embedded inclusion: it rides inside its own document, so it adds no
+      # stage of its own. Its children contribute to the same destination, with
+      # this document appended to their embedded path.
+      #
+      # For Computer.eager_load(port: :device) the :port inclusion emits nothing;
+      # it hands the path [ :port ] to :device, which EmbeddedDistributor then
+      # turns into stages.
+      #
+      # @api private
+      class EmbeddedInclusion < AssociationInclusion
+        class << self
+          # @param [ Mongoid::Association::Relatable ] association The inclusion.
+          #
+          # @return [ true | false ] Whether the association is embedded.
+          def for?(association)
+            association.embedded?
+          end
+        end
+
+        # Add no stage of its own; hand this document down the embedded path so the
+        # children distribute onto it.
+        #
+        # @param [ Array<Hash> ] destination The pipeline (or sub-pipeline) the
+        #   stages are appended to.
+        # @param [ Array<Mongoid::Association::Relatable> ] chain The embedded path
+        #   accumulated from the ancestors above this inclusion (empty at the top).
+        def contribute(destination, chain)
+          @children.each { |child| child.contribute(destination, chain + [ @association ]) }
+        end
+      end
+
+      # A polymorphic inclusion: its target collection varies per document, so it
+      # can't be a $lookup. It adds nothing here; PolymorphicPreloader resolves it
+      # after the roots are materialized.
+      #
+      # @api private
+      class DeferredInclusion < AssociationInclusion
+        class << self
+          # @param [ Mongoid::Association::Relatable ] association The inclusion.
+          #
+          # @return [ true | false ] Whether the association is polymorphic.
+          def for?(association)
+            association.polymorphic?
+          end
+        end
+
+        # Add nothing; PolymorphicPreloader resolves the association after the
+        # roots are materialized.
+        #
+        # @param [ Array<Hash> ] destination The pipeline (unused).
+        # @param [ Array<Mongoid::Association::Relatable> ] chain The embedded path (unused).
+        def contribute(destination, chain); end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/association/eager_load/inclusion_tree.rb b/lib/mongoid/association/eager_load/inclusion_tree.rb
new file mode 100644
index 0000000000..3a25ff770c
--- /dev/null
+++ b/lib/mongoid/association/eager_load/inclusion_tree.rb
@@ -0,0 +1,96 @@
+# frozen_string_literal: true
+
+require 'mongoid/association/eager_load/discriminated_inclusion'
+require 'mongoid/association/eager_load/inclusion'
+
+module Mongoid
+  module Association
+    module EagerLoad
+      # The tree of nested inclusions an eager load asks to load. Built from the
+      # criteria's inclusions, it contributes each root node's stages to the
+      # pipeline, with each node already carrying its own children.
+      #
+      # Each root branch is built from its own copy of the inclusions, and an
+      # inclusion is removed as it is placed, so it lands once per branch even if
+      # more than one parent in that branch points at it, and a circular chain of
+      # inclusions can't loop forever.
+      #
+      # @api private
+      class InclusionTree
+        class << self
+          # Builds the tree for the criteria's inclusions.
+          #
+          # @param [ Array<Mongoid::Association::Relatable> ] inclusions The inclusions.
+          # @param [ LookupPipeline ] pipeline The pipeline being built.
+          #
+          # @return [ InclusionTree ] The tree.
+          def from(inclusions, pipeline)
+            new(inclusions, pipeline, inclusions.to_h { |association| [ association.name, association ] })
+          end
+
+          private :new
+        end
+
+        def initialize(inclusions, pipeline, by_name)
+          @inclusions = inclusions
+          @pipeline = pipeline
+          @by_name = by_name
+        end
+
+        # Contribute each root inclusion's stages to the pipeline. Each root carries
+        # its own children, so the whole tree is appended by recursion from the
+        # roots downward.
+        #
+        # @param [ Array<Hash> ] destination The pipeline the stages are appended to.
+        def contribute_to(destination)
+          roots.each { |root| root.contribute(destination, []) }
+        end
+
+        private
+
+        # A name that more than one subclass defines (with different targets) can't
+        # share one $lookup field, so its nodes -- each carrying its own children --
+        # are grouped and routed by the discriminator instead of becoming separate,
+        # overwriting roots.
+        def roots
+          top_level.group_by(&:name).map do |_name, associations|
+            nodes = associations.map { |association| node(association, @inclusions.dup) }
+            nodes.one? ? nodes.first : DiscriminatedInclusion.new(nodes)
+          end
+        end
+
+        # The inclusions that no other inclusion is the parent of.
+        def top_level
+          @inclusions.reject do |association|
+            association.parent_inclusions.any? { |name| @by_name.key?(name) }
+          end
+        end
+
+        def node(association, available)
+          children = take_children(association, available).map { |child| node(child, available) }
+          AssociationInclusion.for(association, @pipeline, children)
+        end
+
+        # The still-available inclusions parented to +association+, removed as they
+        # are taken so each lands once on this branch. A child belongs here when it
+        # names this association as its parent and its owner shares the target's
+        # class hierarchy, which tells apart children of two unrelated subclasses
+        # that share an association name.
+        def take_children(association, available)
+          children = available.select do |candidate|
+            candidate.parent_inclusions.include?(association.name) &&
+              same_hierarchy?(association.klass, candidate.inverse_class)
+          end
+          available.reject! { |candidate| children.include?(candidate) }
+          children
+        end
+
+        # Whether two classes belong to the same inheritance chain (one is the
+        # other, an ancestor of it, or a descendant of it).
+        def same_hierarchy?(one, other)
+          one <= other || other <= one
+        end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/association/eager_load/lookup_pipeline.rb b/lib/mongoid/association/eager_load/lookup_pipeline.rb
new file mode 100644
index 0000000000..04a4e0760b
--- /dev/null
+++ b/lib/mongoid/association/eager_load/lookup_pipeline.rb
@@ -0,0 +1,121 @@
+# frozen_string_literal: true
+
+require 'mongoid/association/eager_load/embedded_distributor'
+require 'mongoid/association/eager_load/inclusion_tree'
+
+module Mongoid
+  module Association
+    module EagerLoad
+      # Builds the aggregation pipeline that eager-loads a criteria's inclusions
+      # with $lookup.
+      #
+      # It starts with the criteria's own match/sort/skip/limit, then lets each
+      # root of the inclusion tree contribute its stages. This object owns the
+      # stage-building helpers; how each inclusion contributes is the inclusion's
+      # own business (see Inclusion).
+      #
+      # For Band.eager_load(albums: :tracks) the result is roughly:
+      #
+      #   [ <criteria match / sort / skip / limit>,
+      #     { '$lookup' => {                 # JoinedInclusion(:albums)
+      #       'from' => 'albums',
+      #       'localField' => '_id',
+      #       'foreignField' => 'band_id',
+      #       'as' => 'albums',
+      #       'pipeline' => [
+      #         { '$sort' => {
+      #           '_id' => 1
+      #         } },
+      #         { '$lookup' => {             # JoinedInclusion(:tracks), nested
+      #           'as' => 'tracks',
+      #           'pipeline' => [
+      #             { '$sort' => {
+      #               '_id' => 1
+      #             } }
+      #           ]
+      #         } }
+      #       ]
+      #     } } ]
+      #
+      # @api private
+      class LookupPipeline
+        def initialize(criteria)
+          @criteria = criteria
+        end
+
+        # @return [ Array<Hash> ] The aggregation pipeline stages.
+        def stages
+          pipeline = @criteria.selector.to_pipeline
+          pipeline.concat(@criteria.options.to_pipeline_for_lookup)
+          InclusionTree.from(@criteria.inclusions, self).contribute_to(pipeline)
+          pipeline
+        end
+
+        # The $lookup stage for a referenced inclusion: its key fields, a
+        # discriminator match when the target shares its collection with sibling
+        # subclasses, and an order. Children are added by the inclusion itself.
+        #
+        # @param [ Mongoid::Association::Relatable ] association The inclusion.
+        #
+        # @return [ Hash ] The $lookup stage.
+        def lookup_stage_for(association)
+          local_field, foreign_field = lookup_fields(association)
+          sub_pipeline = []
+          sub_pipeline << discriminator_match(association) if association.klass.hereditary?
+          sub_pipeline << order(association)
+          { '$lookup' => {
+            'from' => association.klass.collection.name,
+            'localField' => local_field,
+            'foreignField' => foreign_field,
+            'as' => association.name.to_s,
+            'pipeline' => sub_pipeline
+          } }
+        end
+
+        # Builds the stages that distribute a referenced inclusion living inside an
+        # embedded document onto that document.
+        #
+        # @param [ Mongoid::Association::Relatable ] association The inclusion.
+        # @param [ Array<Mongoid::Association::Relatable> ] chain The embedded path.
+        # @param [ Hash ] lookup_stage The $lookup stage for the association.
+        #
+        # @return [ Array<Hash> ] The stages to append.
+        def distribute(association, chain, lookup_stage)
+          EmbeddedDistributor.for(association: association, chain: chain, lookup_stage: lookup_stage).stages
+        end
+
+        private
+
+        # When the association stores the foreign key on the current document
+        # (belongs_to, has_and_belongs_to_many) the local field is that key; for
+        # the others (has_many, has_one) the key is on the related document.
+        def lookup_fields(association)
+          if association.stores_foreign_key?
+            [ association.foreign_key, association.primary_key ]
+          else
+            [ association.primary_key, association.foreign_key ]
+          end
+        end
+
+        def discriminator_match(association)
+          target = association.klass
+          { '$match' => {
+            target.discriminator_key => { '$in' => target._types }
+          } }
+        end
+
+        def order(association)
+          unless association.order
+            return { '$sort' => {
+              '_id' => 1
+            } }
+          end
+
+          ordering = association.order
+          ordering = { ordering => 1 } unless ordering.is_a?(Hash)
+          { '$sort' => ordering }
+        end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/association/eager_load/polymorphic_preloader.rb b/lib/mongoid/association/eager_load/polymorphic_preloader.rb
new file mode 100644
index 0000000000..d7da8dbfe6
--- /dev/null
+++ b/lib/mongoid/association/eager_load/polymorphic_preloader.rb
@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+
+require 'mongoid/association/eager_load/polymorphic_targets'
+
+module Mongoid
+  module Association
+    module EagerLoad
+      # Resolves a polymorphic belongs_to onto already-materialized root documents.
+      #
+      # A polymorphic belongs_to can't be expressed as a $lookup: its target
+      # collection varies per document. So once the roots are materialized, the
+      # foreign keys are grouped by type, PolymorphicTargets resolves the documents
+      # for those keys, and the result is set on each document.
+      #
+      # @api private
+      class PolymorphicPreloader
+        def initialize(association, root_class)
+          @association = association
+          @root_class = root_class
+        end
+
+        # Resolve and assign the polymorphic target on each of the documents.
+        #
+        # @param [ Array<Mongoid::Document> ] documents The materialized root documents.
+        def preload_into(documents)
+          targets = PolymorphicTargets.for(@association, keys_by_type(documents), @root_class)
+          assign(documents, targets)
+        end
+
+        private
+
+        # The foreign keys on the documents grouped by polymorphic type,
+        # e.g. { "Printer" => [ id1 ], "Scanner" => [ id2 ] }.
+        def keys_by_type(documents)
+          documents.each_with_object({}) do |document, grouped|
+            type, key = reference_on(document)
+            (grouped[type] ||= []) << key if type && key
+          end
+        end
+
+        def assign(documents, targets)
+          documents.each do |document|
+            type, key = reference_on(document)
+            target = targets.dig(type, key) if type && key
+            document.set_relation(@association.name, target)
+          end
+        end
+
+        # The [ type, key ] reference stored on the document for this association,
+        # or an empty pair when the document holds no such reference.
+        def reference_on(document)
+          type_field = @association.inverse_type
+          key_field = @association.foreign_key
+          return [] unless document.respond_to?(type_field) && document.respond_to?(key_field)
+
+          [ document.send(type_field), document.send(key_field) ]
+        end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/association/eager_load/polymorphic_targets.rb b/lib/mongoid/association/eager_load/polymorphic_targets.rb
new file mode 100644
index 0000000000..9459cd02e2
--- /dev/null
+++ b/lib/mongoid/association/eager_load/polymorphic_targets.rb
@@ -0,0 +1,151 @@
+# frozen_string_literal: true
+
+module Mongoid
+  module Association
+    module EagerLoad
+      # The targets of a polymorphic belongs_to, indexed as
+      # { type => { primary_key => document } }. Each subclass reaches the types that
+      # live in one place (the root's database or elsewhere); .for resolves the whole
+      # set, routing each type to the subclass that can reach it.
+      #
+      # @api private
+      class PolymorphicTargets
+        class << self
+          # Resolve every polymorphic target for the foreign keys grouped by type.
+          # The types whose documents share the root's database are fetched together
+          # in one $facet; those living elsewhere are read through their own models.
+          #
+          # @param [ Mongoid::Association::Relatable ] association The polymorphic inclusion.
+          # @param [ Hash ] keys_by_type The foreign keys grouped by type.
+          # @param [ Class ] root_class The class being queried.
+          #
+          # @return [ Hash ] The targets, as { type => { primary_key => document } }.
+          def for(association, keys_by_type, root_class)
+            here, elsewhere = keys_by_type.partition do |type, _keys|
+              in_root_database?(association, type, root_class)
+            end
+            same_database = SameDatabaseTargets.new(association, here.to_h, root_class)
+            other_databases = OtherDatabaseTargets.new(association, elsewhere.to_h)
+            same_database.fetch.merge(other_databases.fetch)
+          end
+
+          private
+
+          # Whether the type's model shares the root's database (and client): exactly
+          # what a $lookup from the root collection can reach.
+          def in_root_database?(association, type, root_class)
+            model = association.resolver.model_for(type)
+            model.client_name == root_class.client_name &&
+              model.database_name == root_class.database_name
+          end
+        end
+
+        def initialize(association, keys_by_type)
+          @association = association
+          @keys_by_type = keys_by_type
+        end
+
+        # @return [ Hash ] The targets, as { type => { primary_key => document } }.
+        def fetch
+          raise NotImplementedError
+        end
+
+        private
+
+        def primary_key
+          @association.primary_key
+        end
+
+        def model_for(type)
+          @association.resolver.model_for(type)
+        end
+
+        # The raw documents instantiated and indexed by their primary key.
+        def indexed(documents, model)
+          documents.map { |document| Factory.from_db(model, document) }
+                   .index_by { |document| document.send(primary_key) }
+        end
+      end
+
+      # Targets that live in the root's own database. A $lookup can reach them, so
+      # every type is fetched together in one $facet aggregation against the root
+      # collection.
+      #
+      # For { 'Printer' => [ id1 ], 'Scanner' => [ id2 ] } it runs:
+      #
+      #   [
+      #     { '$limit' => 1 },   # one input doc, so each facet branch runs once
+      #     { '$facet' => {      # run one $lookup per type within a single query
+      #       'Printer' => [ { '$lookup' => { 'from' => 'printers', ... } }, ... ],
+      #       'Scanner' => [ { '$lookup' => { 'from' => 'scanners', ... } }, ... ]
+      #     } }
+      #   ]
+      #
+      # @api private
+      class SameDatabaseTargets < PolymorphicTargets
+        def initialize(association, keys_by_type, root_class)
+          super(association, keys_by_type)
+          @root_class = root_class
+        end
+
+        # @return [ Hash ] The targets, as { type => { primary_key => document } }.
+        def fetch
+          return {} if @keys_by_type.empty?
+
+          aggregated = @root_class.collection.aggregate([ { '$limit' => 1 }, { '$facet' => facets } ]).first
+          aggregated.to_h do |type, branch|
+            # $limit => 1 makes each branch yield a single wrapper holding the matches.
+            [ type, indexed(branch.first['matches'], model_for(type)) ]
+          end
+        end
+
+        private
+
+        def facets
+          @keys_by_type.to_h do |type, keys|
+            [ type, branch_for(model_for(type).collection.name, keys) ]
+          end
+        end
+
+        # One $facet branch: the documents in +collection_name+ whose primary key
+        # is among +keys+, exposed under "matches".
+        def branch_for(collection_name, keys)
+          [
+            { '$lookup' => {
+              'from' => collection_name,
+              'pipeline' => [
+                { '$match' => {
+                  primary_key => { '$in' => keys.uniq }
+                } }
+              ],
+              'as' => 'matches'
+            } },
+            { '$project' => {
+              '_id' => 0,
+              'matches' => 1
+            } }
+          ]
+        end
+      end
+
+      # Targets kept in another database (or cluster), which a $lookup cannot
+      # reach. Each type is read directly through its own model, which connects
+      # with that model's client.
+      #
+      # For { 'Scanner' => [ id2 ] } it runs, on the Scanner model's own client:
+      #   scanners.find('_id' => { '$in' => [ id2 ] })
+      #
+      # @api private
+      class OtherDatabaseTargets < PolymorphicTargets
+        # @return [ Hash ] The targets, as { type => { primary_key => document } }.
+        def fetch
+          @keys_by_type.to_h do |type, keys|
+            model = model_for(type)
+            documents = model.collection.find(primary_key => { '$in' => keys.uniq })
+            [ type, indexed(documents, model) ]
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/association/eager_loadable.rb b/lib/mongoid/association/eager_loadable.rb
index d46d4c3f44..0dd1bc1535 100644
--- a/lib/mongoid/association/eager_loadable.rb
+++ b/lib/mongoid/association/eager_loadable.rb
@@ -1,6 +1,9 @@
 # frozen_string_literal: true
 
 require 'mongoid/association/eager'
+require 'mongoid/association/eager_load/embedded_distributor'
+require 'mongoid/association/eager_load/polymorphic_preloader'
+require 'mongoid/association/eager_load/lookup_pipeline'
 
 module Mongoid
   module Association
@@ -27,38 +30,49 @@ def eager_load(docs)
 
       # Load the associations for the given documents using $lookup.
       #
-      # If any of the associated collections reside in a different cluster than
-      # the root class, falls back to the #includes behavior and logs a warning.
+      # If any of the associated collections reside in a different cluster or
+      # database than the root class, falls back to the #includes behavior and
+      # logs a warning.
       #
       # @return [ Array<Mongoid::Document> ] The given documents.
       def eager_load_with_lookup
-        offenders = cross_cluster_inclusions
+        offenders = inclusions_unreachable_by_lookup
         if offenders.any?
-          root_client = klass.client_name
-          offender_list = offenders.map { |a| "#{a.name} (#{a.klass.client_name})" }.join(', ')
+          offender_descriptions = offenders.map do |offender|
+            "#{offender.name} (client: #{offender.klass.client_name}, database: #{offender.klass.database_name})"
+          end
           Mongoid.logger.warn(
             'eager_load cannot use $lookup aggregation because the following associations ' \
-            "reside in a different cluster than #{klass} (client: #{root_client}): " \
-            "#{offender_list}. Falling back to #includes behavior."
+            "reside in a different cluster or database than #{klass} " \
+            "(client: #{klass.client_name}, database: #{klass.database_name}): " \
+            "#{offender_descriptions.join(', ')}. Falling back to #includes behavior."
           )
           return eager_load(docs_for_lookup_fallback)
         end
 
-        through_inclusions = criteria.inclusions.select do |assoc|
-          assoc.is_a?(Association::Referenced::HasOneThrough) ||
-            assoc.is_a?(Association::Referenced::HasManyThrough)
+        through_inclusions = criteria.inclusions.select do |association|
+          association.is_a?(Association::Referenced::HasOneThrough) ||
+            association.is_a?(Association::Referenced::HasManyThrough)
         end
 
         if through_inclusions.any?
-          names = through_inclusions.map { |a| ":#{a.name}" }.join(', ')
+          through_names = through_inclusions.map { |association| ":#{association.name}" }
           Mongoid.logger.warn(
-            "#{names} are :through associations and do not support $lookup-based eager " \
-            'loading. All inclusions for this query will be preloaded using separate queries.'
+            "#{through_names.join(', ')} are :through associations and do not support " \
+            '$lookup-based eager loading. All inclusions for this query will be preloaded ' \
+            'using separate queries.'
           )
           return eager_load(docs_for_lookup_fallback)
         end
 
-        preload_for_lookup(criteria)
+        documents = preload_for_lookup(criteria)
+        # A polymorphic belongs_to cannot be expressed as a $lookup: its target
+        # collection varies per document. It is resolved after the roots are
+        # materialized, each inclusion by its own preloader.
+        criteria.inclusions.select(&:polymorphic?).each do |association|
+          EagerLoad::PolymorphicPreloader.new(association, klass).preload_into(documents)
+        end
+        documents
       end
 
       # Load the associations for the given documents. This will be done
@@ -96,39 +110,16 @@ def preload(associations, docs)
         end
       end
 
-      # Load the associations for the given documents. This will be done
-      # recursively to load the associations of the given documents'
-      # associated documents.
+      # Materialize the root documents with their inclusions eager-loaded by a
+      # single $lookup aggregation. The pipeline is built by LookupPipeline; the
+      # polymorphic inclusions it leaves out are resolved by the caller.
       #
-      # @param [ Array<Mongoid::Association::Relatable> ] associations
-      #   The associations to load.
-      # @param [ Array<Mongoid::Document> ] docs The documents.
+      # @param [ Mongoid::Criteria ] criteria The criteria to load.
+      #
+      # @return [ Array<Mongoid::Document> ] The materialized root documents.
       def preload_for_lookup(criteria)
-        assoc_map = criteria.inclusions.group_by(&:inverse_class_name)
-
-        # match first
-        pipeline = criteria.selector.to_pipeline
-        # then sort, skip, limit
-        pipeline.concat(criteria.options.to_pipeline_for_lookup)
-
-        # account for single-collection inheritance
-        root_class = klass.root_class
-
-        if assoc_map[klass.to_s]
-          assoc_map[klass.to_s].each do |assoc|
-            # Create a copy of the mapping for each top-level association to avoid mutation issues
-            pipeline << create_pipeline(assoc, assoc_map.dup)
-          end
-        end
-
-        if klass != root_class && assoc_map[root_class.to_s]
-          assoc_map[root_class.to_s].each do |assoc|
-            # Create a copy of the mapping for each top-level association to avoid mutation issues
-            pipeline << create_pipeline(assoc, assoc_map.dup)
-          end
-        end
-
-        Eager.new(criteria.inclusions, [], true, pipeline).run
+        pipeline = EagerLoad::LookupPipeline.new(criteria).stages
+        Eager.run(criteria.inclusions, [], true, pipeline)
       end
 
       private
@@ -142,71 +133,23 @@ def docs_for_lookup_fallback
         raise NotImplementedError, "#{self.class} must implement #docs_for_lookup_fallback"
       end
 
-      # Returns the inclusions whose target class resides in a different cluster
-      # than the root class.
+      # Returns the inclusions whose target class can't be reached by a $lookup
+      # from the root class, which joins only within the same client and database.
       #
       # @return [ Array<Mongoid::Association::Relatable> ] The offending inclusions.
-      def cross_cluster_inclusions
-        root_client = klass.client_name
-        criteria.inclusions.reject { |assoc| assoc.klass.client_name == root_client }
-      end
-
-      public
-
-      def switch_local_and_foreign_fields?(association)
-        association.is_a?(Mongoid::Association::Referenced::BelongsTo) ||
-          association.is_a?(Mongoid::Association::Referenced::HasAndBelongsToMany)
-      end
-
-      def create_pipeline(current_assoc, mapping)
-        # Build nested pipeline for children and ordering
-        pipeline_stages = []
-
-        # For belongs_to and has_and_belongs_to_many, the foreign key is on the current document
-        # For has_many/has_one, the foreign key is on the related document
-        if switch_local_and_foreign_fields?(current_assoc)
-          local_field = current_assoc.foreign_key
-          foreign_field = current_assoc.primary_key
-        else
-          local_field = current_assoc.primary_key
-          foreign_field = current_assoc.foreign_key
+      def inclusions_unreachable_by_lookup
+        # Polymorphic associations have no single resolvable klass and are not
+        # loaded via $lookup, so they are never offenders.
+        criteria.inclusions.reject do |association|
+          association.polymorphic? || reachable_by_lookup?(association.klass)
         end
+      end
 
-        # Build the 'as' field with embedded path prefix if needed
-        as_field = current_assoc.name.to_s
-
-        stage = {
-          '$lookup' => {
-            'from' => current_assoc.klass.collection.name,
-            'localField' => local_field,
-            'foreignField' => foreign_field,
-            'as' => as_field
-          }
-        }
-
-        # Add ordering if defined on the association, or default to _id for consistent order
-        if current_assoc.order
-          sort_spec = current_assoc.order.is_a?(Hash) ? current_assoc.order : { current_assoc.order => 1 }
-          pipeline_stages << { '$sort' => sort_spec }
-        else
-          # Default to sorting by _id to maintain insertion order consistency
-          pipeline_stages << { '$sort' => { '_id' => 1 } }
-        end
-
-        # Add nested lookups for child associations
-        # Child associations don't need the embedded_path prefix since they're referenced from the looked-up document
-        # Remove this class from the mapping to prevent infinite loops with circular references
-        class_name = current_assoc.klass.to_s
-        if child_assocs = mapping.delete(class_name)
-          child_assocs.each do |child|
-            pipeline_stages << create_pipeline(child, mapping)
-          end
-        end
-
-        # Always add pipeline since we always have at least $sort
-        stage['$lookup']['pipeline'] = pipeline_stages
-
-        stage
+      # Whether a $lookup from a query on the root class can reach the model: it
+      # must live in the same client and database.
+      def reachable_by_lookup?(model)
+        model.client_name == klass.client_name &&
+          model.database_name == klass.database_name
       end
     end
   end
diff --git a/lib/mongoid/association/relatable.rb b/lib/mongoid/association/relatable.rb
index 97b3a18df6..011822eb28 100644
--- a/lib/mongoid/association/relatable.rb
+++ b/lib/mongoid/association/relatable.rb
@@ -342,6 +342,15 @@ def in_to?
         [ Referenced::BelongsTo, Embedded::EmbeddedIn ].any? { |a| is_a?(a) }
       end
 
+      # Is this association a many-to-many association? Such an association stores
+      # its foreign keys as an array on the document itself.
+      #
+      # @return [ true | false ] true if it is a has_and_belongs_to_many
+      #   association, false if not.
+      def many_to_many?
+        is_a?(Referenced::HasAndBelongsToMany)
+      end
+
       private
 
       # Gets the model classes with inverse associations of this model. This is used to determine
diff --git a/lib/mongoid/criteria/includable.rb b/lib/mongoid/criteria/includable.rb
index c5cbe5c408..3b55b84840 100644
--- a/lib/mongoid/criteria/includable.rb
+++ b/lib/mongoid/criteria/includable.rb
@@ -44,11 +44,13 @@ def eager_load(*relations)
         clone
       end
 
-      # Returns whether to use $lookup aggregation for eager loading.
+      # Returns whether to use $lookup aggregation for eager loading. Only when
+      # eager_load was requested and there is something to load: an empty
+      # inclusion list (e.g. eager_load([])) falls back to the normal path.
       #
       # @return [ true | false ] Whether to use $lookup.
       def use_lookup?
-        !!@use_lookup
+        !!@use_lookup && inclusions.any?
       end
 
       # Get a list of criteria that are to be executed for eager loading.
@@ -96,20 +98,42 @@ def add_inclusion(association, parent = nil)
       #   The names of the association(s) to eager load.
       def extract_includes_list(_parent_class, parent, is_eager_load = false, *relations_list)
         relations_list.flatten.each do |relation_object|
-          if relation_object.is_a?(Hash)
-            relation_object.each do |relation, _includes|
-              association = _parent_class.reflect_on_association(relation)
-              raise_eager_error(is_eager_load, _klass, relation) unless association
+          # Normalize a bare association name to a hash with no nested
+          # inclusions, so both forms share one resolution path below.
+          relations = relation_object.is_a?(Hash) ? relation_object : { relation_object => nil }
+
+          relations.each do |relation, nested|
+            associations = resolve_inclusion_associations(_parent_class, relation, is_eager_load)
+            raise_eager_error(is_eager_load, _parent_class, relation) if associations.empty?
+
+            associations.each do |association|
               add_inclusion(association, parent)
-              extract_includes_list(association.klass, association.name, is_eager_load, _includes)
+              extract_includes_list(association.klass, association.name, is_eager_load, nested) if nested
             end
-          else
-            association = _parent_class.reflect_on_association(relation_object)
-            raise_eager_error(is_eager_load, _parent_class, relation_object) unless association
-            add_inclusion(association, parent)
           end
         end
       end
+
+      # Resolve the association(s) matching the given relation name. For the
+      # regular #includes path, only the parent class is consulted. For the
+      # #eager_load ($lookup) path, its subclasses are consulted as well, so
+      # associations defined only on a subclass can also be eager-loaded when
+      # querying through the superclass.
+      #
+      # @param [ Class ] parent_class The class to start the lookup from.
+      # @param [ Symbol | String ] relation The association name.
+      # @param [ Boolean ] is_eager_load Whether to consider subclasses.
+      #
+      # @return [ Array<Mongoid::Association::Relatable> ] Matching associations.
+      def resolve_inclusion_associations(parent_class, relation, is_eager_load)
+        if association = parent_class.reflect_on_association(relation)
+          return [ association ]
+        end
+
+        return [] unless is_eager_load
+
+        parent_class.descendants.filter_map { |sub| sub.reflect_on_association(relation) }
+      end
     end
 
     def raise_eager_error(is_eager_load, klass, relation)
diff --git a/spec/mongoid/association/eager_spec.rb b/spec/mongoid/association/eager_spec.rb
index c4dafc4b0c..2ea09771a7 100644
--- a/spec/mongoid/association/eager_spec.rb
+++ b/spec/mongoid/association/eager_spec.rb
@@ -679,9 +679,9 @@ class Voucher
         expect(docs.first.posts).to eq([ post ])
       end
 
-      it 'mentions the offending association and client in the warning' do
+      it 'mentions the offending association, client, and database in the warning' do
         expect(Mongoid.logger).to receive(:warn).with(
-          /posts \(other_cluster\)/
+          /posts \(client: other_cluster, database: /
         )
         context.eager_load_with_lookup
       end
diff --git a/spec/mongoid/criteria/includable_spec.rb b/spec/mongoid/criteria/includable_spec.rb
index 66f7476a3f..31404350bd 100644
--- a/spec/mongoid/criteria/includable_spec.rb
+++ b/spec/mongoid/criteria/includable_spec.rb
@@ -1414,6 +1414,881 @@ class C
     # 4.4 lookup does not support the lookup pipeline as it is currently written
     min_server_version '5.0'
     it_behaves_like 'eager loading', :eager_load
+
+    context 'when the inclusion list is empty' do
+      let!(:band) { Band.create! }
+
+      it 'returns the documents without eager loading' do
+        expect(Band.eager_load([]).first).to eq(band)
+      end
+    end
+
+    context 'when the association is defined on a subclass of the queried class' do
+      before(:all) do
+        class Machine
+          include Mongoid::Document
+        end
+
+        class Peripheral
+          include Mongoid::Document
+        end
+
+        class Webcam < Peripheral
+        end
+
+        class Mouse < Peripheral
+          belongs_to :machine
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Mouse)
+        Object.send(:remove_const, :Webcam)
+        Object.send(:remove_const, :Peripheral)
+        Object.send(:remove_const, :Machine)
+      end
+
+      let!(:machine) { Machine.create! }
+      let!(:webcam) { Webcam.create! }
+      let!(:mouse) { Mouse.create!(machine: machine) }
+
+      it 'eager-loads it through the queried superclass' do
+        expect_query(1) do
+          loaded = Peripheral.eager_load(:machine).detect { |doc| doc.is_a?(Mouse) }
+          expect(loaded.machine).to eq(machine)
+        end
+      end
+    end
+
+    context 'when the association is defined on a subclass of an associated class' do
+      before(:all) do
+        class Brand
+          include Mongoid::Document
+        end
+
+        class Machine
+          include Mongoid::Document
+        end
+
+        class Laptop < Machine
+          belongs_to :brand
+        end
+
+        class Peripheral
+          include Mongoid::Document
+        end
+
+        class Mouse < Peripheral
+          belongs_to :machine
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Mouse)
+        Object.send(:remove_const, :Peripheral)
+        Object.send(:remove_const, :Laptop)
+        Object.send(:remove_const, :Machine)
+        Object.send(:remove_const, :Brand)
+      end
+
+      let!(:brand) { Brand.create! }
+      let!(:laptop) { Laptop.create!(brand: brand) }
+      let!(:mouse) { Mouse.create!(machine: laptop) }
+
+      it 'eager-loads it through the parent association' do
+        expect_query(1) do
+          loaded = Peripheral.eager_load(machine: :brand).first
+          expect(loaded.machine).to eq(laptop)
+          expect(loaded.machine.brand).to eq(brand)
+        end
+      end
+    end
+
+    context 'when a has_many is defined only on a subclass' do
+      before(:all) do
+        class Gadget
+          include Mongoid::Document
+        end
+
+        class Speaker < Gadget
+          has_many :cables
+        end
+
+        class Cable
+          include Mongoid::Document
+
+          belongs_to :speaker
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Cable)
+        Object.send(:remove_const, :Speaker)
+        Object.send(:remove_const, :Gadget)
+      end
+
+      let!(:speaker) { Speaker.create! }
+      let!(:cable) { Cable.create!(speaker: speaker) }
+
+      it 'eager-loads it through the queried superclass' do
+        expect_query(1) do
+          loaded = Gadget.eager_load(:cables).first
+          expect(loaded.cables).to eq([ cable ])
+        end
+      end
+    end
+
+    context 'when a has_one is defined only on a subclass' do
+      before(:all) do
+        class Gadget
+          include Mongoid::Document
+        end
+
+        class Speaker < Gadget
+          has_one :cable
+        end
+
+        class Cable
+          include Mongoid::Document
+
+          belongs_to :speaker
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Cable)
+        Object.send(:remove_const, :Speaker)
+        Object.send(:remove_const, :Gadget)
+      end
+
+      let!(:speaker) { Speaker.create! }
+      let!(:cable) { Cable.create!(speaker: speaker) }
+
+      it 'eager-loads it through the queried superclass' do
+        expect_query(1) do
+          loaded = Gadget.eager_load(:cable).first
+          expect(loaded.cable).to eq(cable)
+        end
+      end
+    end
+
+    context 'when a has_and_belongs_to_many is defined only on a subclass' do
+      before(:all) do
+        class Cable
+          include Mongoid::Document
+
+          has_and_belongs_to_many :speakers
+        end
+
+        class Gadget
+          include Mongoid::Document
+        end
+
+        class Speaker < Gadget
+          has_and_belongs_to_many :cables
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Speaker)
+        Object.send(:remove_const, :Gadget)
+        Object.send(:remove_const, :Cable)
+      end
+
+      let!(:cable) { Cable.create! }
+      let!(:speaker) { Speaker.create!(cables: [ cable ]) }
+
+      it 'eager-loads it through the queried superclass' do
+        expect_query(1) do
+          loaded = Gadget.eager_load(:cables).first
+          expect(loaded.cables).to eq([ cable ])
+        end
+      end
+    end
+
+    context 'when a belongs_to lives inside an embedded document' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :computer
+          belongs_to :device
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_one :port
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device) { Device.create! }
+      let!(:computer) { Computer.create!(port: Port.new(device: device)) }
+
+      it 'eager-loads it through the embedded document' do
+        expect_query(1) do
+          loaded = Computer.eager_load(port: :device).first
+          expect(loaded.port.device).to eq(device)
+        end
+      end
+
+      it 'leaves the embedded document absent instead of synthesizing it' do
+        without_port = Computer.create!
+        loaded = expect_query(1) do
+          Computer.eager_load(port: :device).to_a.index_by(&:id)
+        end
+        expect_no_queries do
+          expect(loaded[without_port.id].port).to be_nil
+        end
+      end
+    end
+
+    context 'when two embeds_one associations of the same class are eager-loaded' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :computer
+          belongs_to :device
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_one :left_port, class_name: 'Port'
+          embeds_one :right_port, class_name: 'Port'
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device1) { Device.create! }
+      let!(:device2) { Device.create! }
+      let!(:computer) do
+        Computer.create!(
+          left_port: Port.new(device: device1),
+          right_port: Port.new(device: device2)
+        )
+      end
+
+      it 'eager-loads the belongs_to in both embedded documents' do
+        expect_query(1) do
+          loaded = Computer.eager_load(left_port: :device, right_port: :device).first
+          expect(loaded.left_port.device).to eq(device1)
+          expect(loaded.right_port.device).to eq(device2)
+        end
+      end
+    end
+
+    context 'when a belongs_to lives inside a doubly embedded document' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :panel
+          belongs_to :device
+        end
+
+        class Panel
+          include Mongoid::Document
+
+          embedded_in :computer
+          embeds_one :port
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_one :panel
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Panel)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device) { Device.create! }
+      let!(:computer) { Computer.create!(panel: Panel.new(port: Port.new(device: device))) }
+
+      it 'eager-loads it through both embedded documents' do
+        expect_query(1) do
+          loaded = Computer.eager_load(panel: { port: :device }).first
+          expect(loaded.panel.port.device).to eq(device)
+        end
+      end
+    end
+
+    context 'when a belongs_to is defined only on a subclass of an embedded document' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :computer
+        end
+
+        class UsbPort < Port
+          belongs_to :device
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_one :port
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :UsbPort)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device) { Device.create! }
+      let!(:computer) { Computer.create!(port: UsbPort.new(device: device)) }
+
+      it 'eager-loads it through the embedded subclass' do
+        expect_query(1) do
+          loaded = Computer.eager_load(port: :device).first
+          expect(loaded.port.device).to eq(device)
+        end
+      end
+    end
+
+    context 'when a belongs_to lives inside an embeds_many' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :computer
+          belongs_to :device
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_many :ports
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device1) { Device.create! }
+      let!(:device2) { Device.create! }
+      let!(:computer) { Computer.create!(ports: [ Port.new(device: device1), Port.new(device: device2) ]) }
+
+      it 'eager-loads each embedded belongs_to without extra queries' do
+        expect_query(1) do
+          loaded = Computer.eager_load(ports: :device).first
+          expect(loaded.ports.map(&:device)).to eq([ device1, device2 ])
+        end
+      end
+    end
+
+    context 'when a has_and_belongs_to_many lives inside an embedded document' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :computer
+          has_and_belongs_to_many :devices
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_one :port
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device1) { Device.create! }
+      let!(:device2) { Device.create! }
+      let!(:computer) { Computer.create!(port: Port.new(devices: [ device1, device2 ])) }
+
+      it 'eager-loads the devices referenced by the embedded port' do
+        expect_query(1) do
+          loaded = Computer.eager_load(port: :devices).first
+          expect(loaded.port.devices).to eq([ device1, device2 ])
+        end
+      end
+    end
+
+    context 'when an embedded belongs_to targets a superclass of the queried subclass' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Computer < Device
+          embeds_one :port
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :computer
+          belongs_to :device
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device) { Device.create! }
+      let!(:computer) { Computer.create!(port: Port.new(device: device)) }
+
+      it 'eager-loads the embedded belongs_to through the queried subclass' do
+        expect_query(1) do
+          loaded = Computer.eager_load(port: :device).first
+          expect(loaded.port.device).to eq(device)
+        end
+      end
+    end
+
+    context 'when an embedded association is nested under a referenced one' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :computer
+          belongs_to :device
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_one :port
+        end
+
+        class Desk
+          include Mongoid::Document
+
+          belongs_to :computer
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Desk)
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device) { Device.create! }
+      let!(:computer) { Computer.create!(port: Port.new(device: device)) }
+      let!(:desk) { Desk.create!(computer: computer) }
+
+      it 'eager-loads the embedded association together with its referenced child' do
+        expect_query(1) do
+          loaded = Desk.eager_load(computer: { port: :device }).first
+          expect(loaded.computer.port.device).to eq(device)
+        end
+      end
+    end
+
+    context 'when an embedded association is nested two levels under a referenced one' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Pin
+          include Mongoid::Document
+
+          embedded_in :port
+          belongs_to :device
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :computer
+          embeds_one :pin
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_one :port
+        end
+
+        class Desk
+          include Mongoid::Document
+
+          belongs_to :computer
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Desk)
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Pin)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:device) { Device.create! }
+      let!(:computer) { Computer.create!(port: Port.new(pin: Pin.new(device: device))) }
+      let!(:desk) { Desk.create!(computer: computer) }
+
+      it 'eager-loads the doubly embedded association together with its referenced child' do
+        expect_query(1) do
+          loaded = Desk.eager_load(computer: { port: { pin: :device } }).first
+          expect(loaded.computer.port.pin.device).to eq(device)
+        end
+      end
+    end
+
+    context 'when an embeds_many is nested inside another embeds_many under a referenced one' do
+      before(:all) do
+        class Device
+          include Mongoid::Document
+        end
+
+        class Port
+          include Mongoid::Document
+
+          embedded_in :card
+          belongs_to :device
+        end
+
+        class Card
+          include Mongoid::Document
+
+          embedded_in :computer
+          embeds_many :ports
+        end
+
+        class Computer
+          include Mongoid::Document
+
+          embeds_many :cards
+        end
+
+        class Desk
+          include Mongoid::Document
+
+          belongs_to :computer
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Desk)
+        Object.send(:remove_const, :Computer)
+        Object.send(:remove_const, :Card)
+        Object.send(:remove_const, :Port)
+        Object.send(:remove_const, :Device)
+      end
+
+      let!(:devices) { Array.new(4) { Device.create! } }
+      let!(:computer) do
+        Computer.create!(
+          cards: [
+            Card.new(ports: [ Port.new(device: devices[0]), Port.new(device: devices[1]) ]),
+            Card.new(ports: [ Port.new(device: devices[2]), Port.new(device: devices[3]) ])
+          ]
+        )
+      end
+      let!(:desk) { Desk.create!(computer: computer) }
+
+      it 'eager-loads each deeply embedded belongs_to correlated to its own element' do
+        expect_query(1) do
+          loaded = Desk.eager_load(computer: { cards: { ports: :device } }).first
+          got = loaded.computer.cards.flat_map { |card| card.ports.map(&:device) }
+          expect(got).to eq(devices)
+        end
+      end
+    end
+
+    context 'when a polymorphic belongs_to is eager-loaded' do
+      before(:all) do
+        class Printer
+          include Mongoid::Document
+        end
+
+        class Scanner
+          include Mongoid::Document
+        end
+
+        class Cartridge
+          include Mongoid::Document
+
+          belongs_to :device, polymorphic: true
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Cartridge)
+        Object.send(:remove_const, :Scanner)
+        Object.send(:remove_const, :Printer)
+      end
+
+      let!(:printer) { Printer.create! }
+      let!(:scanner) { Scanner.create! }
+      let!(:printer_cartridge) { Cartridge.create!(device: printer) }
+      let!(:scanner_cartridge) { Cartridge.create!(device: scanner) }
+
+      it 'eager-loads polymorphic targets of different types in a single extra query' do
+        expect_query(2) do
+          loaded = Cartridge.eager_load(:device).to_a.index_by(&:id)
+          expect(loaded[printer_cartridge.id].device).to eq(printer)
+          expect(loaded[scanner_cartridge.id].device).to eq(scanner)
+        end
+      end
+    end
+
+    context 'when a polymorphic target is stored in another database' do
+      before(:all) do
+        class LocalDevice
+          include Mongoid::Document
+        end
+
+        class RemoteDevice
+          include Mongoid::Document
+
+          store_in database: 'secondary'
+        end
+
+        class Accessory
+          include Mongoid::Document
+
+          belongs_to :device, polymorphic: true
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Accessory)
+        Object.send(:remove_const, :RemoteDevice)
+        Object.send(:remove_const, :LocalDevice)
+      end
+
+      # RemoteDevice lives in its own database, which the global cleanup hook
+      # (default database only) does not touch, so drop it here.
+      after { RemoteDevice.collection.drop }
+
+      let!(:local_device) { LocalDevice.create! }
+      let!(:remote_device) { RemoteDevice.create! }
+      let!(:local_accessory) { Accessory.create!(device: local_device) }
+      let!(:remote_accessory) { Accessory.create!(device: remote_device) }
+
+      it 'eager-loads the target stored in the other database' do
+        # One query materializes the roots, one $facet fetches the same-database
+        # target, and one direct query fetches the target in the other database.
+        loaded = expect_query(3) do
+          Accessory.eager_load(:device).to_a.index_by(&:id)
+        end
+
+        expect_no_queries do
+          expect(loaded[local_accessory.id].device).to eq(local_device)
+          expect(loaded[remote_accessory.id].device).to eq(remote_device)
+        end
+      end
+    end
+
+    context 'when a non-polymorphic target is stored in another database' do
+      before(:all) do
+        class RemoteServer
+          include Mongoid::Document
+
+          store_in database: 'secondary'
+        end
+
+        class Workstation
+          include Mongoid::Document
+
+          belongs_to :server, class_name: 'RemoteServer', optional: true
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Workstation)
+        Object.send(:remove_const, :RemoteServer)
+      end
+
+      # RemoteServer lives in its own database, which the global cleanup hook
+      # (default database only) does not touch, so drop it here.
+      after { RemoteServer.collection.drop }
+
+      let!(:server) { RemoteServer.create! }
+      let!(:workstation) { Workstation.create!(server: server) }
+
+      it 'eager-loads the target stored in the other database' do
+        # A $lookup cannot reach another database, so this falls back to
+        # separate-query loading: one query materializes the roots and one fetches
+        # the targets, instead of silently returning nothing.
+        loaded = expect_query(2) do
+          Workstation.eager_load(:server).to_a
+        end
+
+        expect_no_queries do
+          expect(loaded.first.server).to eq(server)
+        end
+      end
+    end
+
+    context 'when a has_many of the same name is defined on more than one subclass' do
+      before(:all) do
+        class Supplier
+          include Mongoid::Document
+        end
+
+        class Cog
+          include Mongoid::Document
+
+          belongs_to :machine, optional: true
+          belongs_to :supplier, optional: true
+        end
+
+        class Belt
+          include Mongoid::Document
+
+          belongs_to :machine, optional: true
+          belongs_to :supplier, optional: true
+        end
+
+        class Machine
+          include Mongoid::Document
+        end
+
+        class Lathe < Machine
+          has_many :widgets, class_name: 'Cog', inverse_of: :machine
+        end
+
+        class Press < Machine
+          has_many :widgets, class_name: 'Belt', inverse_of: :machine
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Press)
+        Object.send(:remove_const, :Lathe)
+        Object.send(:remove_const, :Machine)
+        Object.send(:remove_const, :Belt)
+        Object.send(:remove_const, :Cog)
+        Object.send(:remove_const, :Supplier)
+      end
+
+      let!(:cog_supplier) { Supplier.create! }
+      let!(:belt_supplier) { Supplier.create! }
+      let!(:lathe) { Lathe.create! }
+      let!(:press) { Press.create! }
+      let!(:cog) { Cog.create!(machine: lathe, supplier: cog_supplier) }
+      let!(:belt) { Belt.create!(machine: press, supplier: belt_supplier) }
+
+      it 'loads each subclass association without one overwriting the other' do
+        loaded = expect_query(1) do
+          Machine.eager_load(:widgets).to_a.index_by(&:id)
+        end
+        expect_no_queries do
+          expect(loaded[lathe.id].widgets).to eq([ cog ])
+          expect(loaded[press.id].widgets).to eq([ belt ])
+        end
+      end
+
+      it 'eager-loads inclusions nested under each subclass association' do
+        loaded = expect_query(1) do
+          Machine.eager_load(widgets: :supplier).to_a.index_by(&:id)
+        end
+        expect_no_queries do
+          expect(loaded[lathe.id].widgets.first.supplier).to eq(cog_supplier)
+          expect(loaded[press.id].widgets.first.supplier).to eq(belt_supplier)
+        end
+      end
+    end
+
+    context 'when a has_many targets a class sharing its collection with sibling subclasses' do
+      before(:all) do
+        class Part
+          include Mongoid::Document
+        end
+
+        class Chip < Part
+          belongs_to :board, class_name: 'Board', optional: true
+        end
+
+        class Cable < Part
+          belongs_to :board, class_name: 'Board', optional: true
+        end
+
+        class Board
+          include Mongoid::Document
+
+          has_many :chips, class_name: 'Chip', inverse_of: :board
+        end
+      end
+
+      after(:all) do
+        Object.send(:remove_const, :Board)
+        Object.send(:remove_const, :Cable)
+        Object.send(:remove_const, :Chip)
+        Object.send(:remove_const, :Part)
+      end
+
+      let!(:board) { Board.create! }
+      let!(:chip) { Chip.create!(board: board) }
+      let!(:cable) { Cable.create!(board: board) }
+
+      it 'eager-loads only the chips, not the sibling subclass' do
+        expect_query(1) do
+          loaded = Board.eager_load(:chips).first
+          expect(loaded.chips.to_a).to eq([ chip ])
+        end
+      end
+    end
   end
 
   describe '#inclusions' do