diff --git a/docs/reference/configuration.txt b/docs/reference/configuration.txt
index a018f43086..35827b3d6f 100644
--- a/docs/reference/configuration.txt
+++ b/docs/reference/configuration.txt
@@ -265,6 +265,16 @@ for details on driver options.
       # if the database name is not explicitly defined. (default: nil)
       app_name: MyApplicationName
 
+      # Type of executor for queries scheduled using ``load_async`` method.
+      #
+      # There are two possible values for this option:
+      #
+      #   - :immediate - Queries will be immediately executed on a current thread.
+      #       This is the default option.
+      #   - :global_thread_pool - Queries will be executed asynchronously in
+      #       background using a thread pool.
+      #async_query_executor: :immediate
+
       # Mark belongs_to associations as required by default, so that saving a
       # model with a missing belongs_to association will trigger a validation
       # error. (default: true)
@@ -358,6 +368,11 @@ for details on driver options.
       # Raise an exception when a field is redefined. (default: false)
       duplicate_fields_exception: false
 
+      # Defines how many asynchronous queries can be executed concurrently.
+      # This option should be set only if `async_query_executor` option is set
+      # to `:global_thread_pool`.
+      #global_executor_concurrency: nil
+
       # Include the root model name in json serialization. (default: false)
       include_root_in_json: false
 
diff --git a/docs/reference/queries.txt b/docs/reference/queries.txt
index cdab68baa7..3b837b9594 100644
--- a/docs/reference/queries.txt
+++ b/docs/reference/queries.txt
@@ -1887,8 +1887,8 @@ in the default scope, the value in the default scope takes precedence:
   Band.new # => #<Band _id: 5c3f74ddce4ef3791abbb088, name: nil, active: false>
 
 Because a default scope initializes fields in new models as just described,
-defining a default scope with a dotted key and a simple literal value is
-not possible:
+defining a default scope with a dotted key and a simple literal value, while
+possible, is not recommended:
 
 .. code-block:: ruby
 
@@ -1900,7 +1900,36 @@ not possible:
     default_scope ->{ where('tags.foo' => 'bar') }
   end
 
-  Band.create! # exception: BSON::String::IllegalKey ('tags.foo' is an illegal key in MongoDB. Keys may not start with '$' or contain a '.'.)
+  Band.create!
+  # => Created document: {"_id"=>BSON::ObjectId('632de48f3282a404bee1877b'), "tags.foo"=>"bar"}
+  Band.create!(tags: { 'foo' => 'bar' })
+  # => Created document: {"_id"=>BSON::ObjectId('632de4ad3282a404bee1877c'), "tags.foo"=>"bar", "tags"=>{"foo"=>"bar"}}
+  Band.all.to_a
+  # => [ #<Band _id: 632de4ad3282a404bee1877c, tags: {"foo"=>"bar"}> ]
+
+Mongoid 8 allows dotted keys to be used in Mongoid, and when creating a document,
+the scope is added as a dotted key in the attributes:
+
+.. code-block:: ruby
+
+  Band.new.attribute
+  # => {"_id"=>BSON::ObjectId('632de97d3282a404bee1877d'), "tags.foo"=>"bar"}
+
+Whereas when querying, Mongoid looks for an embedded document:
+
+.. code-block:: ruby
+
+  Band.create!
+  # => Created document: {"_id"=>BSON::ObjectId('632de48f3282a404bee1877b'), "tags.foo"=>"bar"}
+  Band.where
+  # => #<Mongoid::Criteria
+         selector: {"tags.foo"=>"bar"}
+         options:  {}
+         class:    Band
+         embedded: false>
+  # This looks for something like: { tags: { "foo" => "bar" } }
+  Band.count
+  # => 0
 
 A workaround is to define the default scope as a complex query:
 
@@ -1914,9 +1943,11 @@ A workaround is to define the default scope as a complex query:
     default_scope ->{ where('tags.foo' => {'$eq' => 'bar'}) }
   end
 
-  Band.create!(tags: {hello: 'world'})
-  Band.create!(tags: {foo: 'bar'})
-  Band.count # => 1
+  Band.create!(tags: { hello: 'world' })
+  Band.create!(tags: { foo: 'bar' })
+  # does not add a "tags.foo" dotted attribute
+  Band.count
+  # => 1
 
 You can tell Mongoid not to apply the default scope by using
 ``unscoped``, which can be inline or take a block.
@@ -2388,5 +2419,72 @@ will query the database and separately cache their results.
 To use the cached results, call ``all.to_a.first`` on the model class.
 
 
-.. _legacy-query-cache-limitations:
+.. _load-async:
+
+Asynchronous Queries
+====================
+
+Mongoid allows running database queries asynchronously in the background.
+This can be beneficial when there is a need to get documents from different
+collections.
+
+In order to schedule an asynchronous query call the ``load_async`` method on a
+``Criteria``:
+
+.. code-block:: ruby
+
+  class PagesController < ApplicationController
+    def index
+      @active_bands = Band.where(active: true).load_async
+      @best_events = Event.best.load_async
+      @public_articles = Article.where(public: true).load_async
+    end
+  end
+
+In the above example three queries will be scheduled for asynchronous execution.
+Results of the queries can be later accessed as usual:
+
+.. code-block:: html
+
+  <ul>
+    <%- @active_bands.each do -%>
+      <li><%= band.name %></li>
+    <%- end -%>
+  </ul>
+
+Even if a query is scheduled for asynchronous execution, it might be executed
+synchronously on the caller's thread. There are three possible scenarios depending
+on when the query results are being accessed:
+
+#. If the scheduled asynchronous task has been already executed, the results are returned.
+#. If the task has been started, but not finished yet, the caller's thread blocks until the task is finished.
+#. If the task has not been started yet, it is removed from the execution queue, and the query is executed synchronously on the caller's thread.
+
+.. note::
+
+  Even though ``load_async`` method returns a ``Criteria`` object, you should not
+  do any operations on this object except accessing query results. The query is
+  scheduled for execution immediately after calling ``load_async``, therefore
+  later changes to the `Criteria`` object may not be applied.
+
+
+Configuring asynchronous query execution
+----------------------------------------
+
+Asynchronous queries are disabled by default. When asynchronous queries are
+disabled, ``load_async`` will execute the query immediately on the current thread,
+blocking as necessary. Therefore, calling ``load_async`` on criteria in this case
+is roughly the equivalent of calling ``to_a`` to force query execution.
+
+In order to enable asynchronous query execution, the following config options
+must be set:
+
+.. code-block:: yaml
 
+  development:
+    ...
+    options:
+      # Execute asynchronous queries using a global thread pool.
+      async_query_executor: :global_thread_pool
+      # Number of threads in the pool. The default is 4.
+      # global_executor_concurrency: 4
diff --git a/docs/release-notes/mongoid-8.1.txt b/docs/release-notes/mongoid-8.1.txt
index 036f4f0019..a7195e8088 100644
--- a/docs/release-notes/mongoid-8.1.txt
+++ b/docs/release-notes/mongoid-8.1.txt
@@ -17,6 +17,11 @@ The complete list of releases is available `on GitHub
 please consult GitHub releases for detailed release notes and JIRA for
 the complete list of issues fixed in each release, including bug fixes.
 
+Added ``load_async`` method on ``Criteria`` to asynchronously load documents
+----------------------------------------------------------------------------
+
+The new ``load_async`` method on ``Criteria`` allows :ref:`running database queries asynchronously <load-async>`.
+
 
 Added ``attribute_before_last_save``, ``saved_change_to_attribute``, ``saved_change_to_attribute?``, and ``will_save_change_to_attribute?``  methods
 ----------------------------------------------------------------------------------------------------------------------------------------------------
@@ -245,8 +250,8 @@ for more details.
 Added ``readonly!`` method and ``legacy_readonly`` feature flag
 ---------------------------------------------------------------
 
-Mongoid 8.1 changes the meaning of read-only documents. In Mongoid 8.1 with 
-this feature flag turned off, a document becomes read-only when calling the 
+Mongoid 8.1 changes the meaning of read-only documents. In Mongoid 8.1 with
+this feature flag turned off, a document becomes read-only when calling the
 ``readonly!`` method:
 
 .. code:: ruby
@@ -261,8 +266,8 @@ this feature flag turned off, a document becomes read-only when calling the
 With this feature flag turned off, a ``ReadonlyDocument`` error will be
 raised when destroying or deleting, as well as when saving or updating.
 
-Prior to Mongoid 8.1 and in 8.1 with the ``legacy_readonly`` feature flag 
-turned on, documents become read-only when they are projected (i.e. using 
+Prior to Mongoid 8.1 and in 8.1 with the ``legacy_readonly`` feature flag
+turned on, documents become read-only when they are projected (i.e. using
 ``#only`` or ``#without``).
 
 .. code:: ruby
@@ -278,7 +283,7 @@ turned on, documents become read-only when they are projected (i.e. using
   band.destroy # => raises ReadonlyDocument error
 
 Note that with this feature flag on, a ``ReadonlyDocument`` error will only be
-raised when destroying or deleting, and not on saving or updating. See the 
+raised when destroying or deleting, and not on saving or updating. See the
 section on :ref:`Read-only Documents <readonly-documents>` for more details.
 
 
diff --git a/lib/config/locales/en.yml b/lib/config/locales/en.yml
index 922109b312..70d24850fe 100644
--- a/lib/config/locales/en.yml
+++ b/lib/config/locales/en.yml
@@ -133,6 +133,14 @@ en:
             A collation option is only supported if the query is executed on a MongoDB server
             with version >= 3.4."
           resolution: "Remove the collation option from the query."
+        invalid_async_query_executor:
+          message: "Invalid async_query_executor option: %{executor}."
+          summary: "A invalid async query executor was specified.
+            The valid options are: %{options}."
+          resolution: "Pick an allowed option or fix the typo. If you were
+            expecting the option to be there, please consult the following page
+            with respect to Mongoid's configuration:\n\n
+            \_\_https://www.mongodb.com/docs/mongoid/current/reference/configuration/#mongoid-configuration-options"
         invalid_config_file:
           message: "Invalid configuration file: %{path}."
           summary: "Your mongoid.yml configuration file does not contain the
@@ -238,6 +246,13 @@ en:
           resolution: "Please provide a valid type value for the field.
             Refer to:
             https://docs.mongodb.com/mongoid/current/reference/fields/#using-symbols-or-strings-instead-of-classes"
+        invalid_global_executor_concurrency:
+          message: "Invalid global_executor_concurrency option."
+          summary: "You set global_executor_concurrency while async_query_executor
+            option is not set to :global_thread_pool. The global_executor_concurrency is
+            allowed only for the global thread pool executor."
+          resolution: "Set global_executor_concurrency option to :global_thread_pool
+            or remove global_executor_concurrency option."
         invalid_includes:
           message: "Invalid includes directive: %{klass}.includes(%{args})"
           summary: "Eager loading in Mongoid only supports providing arguments
diff --git a/lib/mongoid.rb b/lib/mongoid.rb
index 22816bd9e2..fdea622c97 100644
--- a/lib/mongoid.rb
+++ b/lib/mongoid.rb
@@ -12,6 +12,8 @@
 require "active_support/time_with_zone"
 require "active_model"
 
+require 'concurrent-ruby'
+
 require "mongo"
 require 'mongo/active_support'
 
diff --git a/lib/mongoid/association/referenced/has_and_belongs_to_many.rb b/lib/mongoid/association/referenced/has_and_belongs_to_many.rb
index 1daa94f963..a369fb6b17 100644
--- a/lib/mongoid/association/referenced/has_and_belongs_to_many.rb
+++ b/lib/mongoid/association/referenced/has_and_belongs_to_many.rb
@@ -263,8 +263,12 @@ def with_ordering(criteria)
 
         def query_criteria(id_list)
           crit = relation_class.criteria
-          crit = crit.apply_scope(scope)
-          crit = crit.all_of(primary_key => {"$in" => id_list || []})
+          crit = if id_list
+            crit = crit.apply_scope(scope)
+            crit.all_of(primary_key => { "$in" => id_list })
+          else
+            crit.none
+          end
           with_ordering(crit)
         end
       end
diff --git a/lib/mongoid/config.rb b/lib/mongoid/config.rb
index ef82d3ea1d..de0067351e 100644
--- a/lib/mongoid/config.rb
+++ b/lib/mongoid/config.rb
@@ -127,7 +127,20 @@ module Config
     # always return a Hash.
     option :legacy_attributes, default: false
 
-    # When this flag is false, a document will become read-only only once the 
+    # Sets the async_query_executor for the application. By default the thread pool executor
+    #   is set to `:immediate. Options are:
+    #
+    #   - :immediate - Initializes a single +Concurrent::ImmediateExecutor+
+    #   - :global_thread_pool - Initializes a single +Concurrent::ThreadPoolExecutor+
+    #      that uses the +async_query_concurrency+ for the +max_threads+ value.
+    option :async_query_executor, default: :immediate
+
+    # Defines how many asynchronous queries can be executed concurrently.
+    # This option should be set only if `async_query_executor` is set
+    # to `:global_thread_pool`.
+    option :global_executor_concurrency, default: nil
+
+    # When this flag is false, a document will become read-only only once the
     # #readonly! method is called, and an error will be raised on attempting
     # to save or update such documents, instead of just on delete. When this
     # flag is true, a document is only read-only if it has been projected
@@ -308,6 +321,7 @@ def truncate!
     # @param [ Hash ] options The configuration options.
     def options=(options)
       if options
+        Validators::AsyncQueryExecutor.validate(options)
         options.each_pair do |option, value|
           Validators::Option.validate(option)
           send("#{option}=", value)
diff --git a/lib/mongoid/config/validators.rb b/lib/mongoid/config/validators.rb
index 5fbe21403d..3675bea76f 100644
--- a/lib/mongoid/config/validators.rb
+++ b/lib/mongoid/config/validators.rb
@@ -1,4 +1,5 @@
 # frozen_string_literal: true
 
+require "mongoid/config/validators/async_query_executor"
 require "mongoid/config/validators/option"
 require "mongoid/config/validators/client"
diff --git a/lib/mongoid/config/validators/async_query_executor.rb b/lib/mongoid/config/validators/async_query_executor.rb
new file mode 100644
index 0000000000..b1721b5af4
--- /dev/null
+++ b/lib/mongoid/config/validators/async_query_executor.rb
@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+module Mongoid
+  module Config
+    module Validators
+
+      # Validator for async query executor configuration.
+      #
+      # @api private
+      module AsyncQueryExecutor
+        extend self
+
+
+        def validate(options)
+          if options.key?(:async_query_executor)
+            if options[:async_query_executor].to_sym == :immediate && !options[:global_executor_concurrency].nil?
+              raise Errors::InvalidGlobalExecutorConcurrency
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/contextual.rb b/lib/mongoid/contextual.rb
index 6d5829f791..648ccae1f6 100644
--- a/lib/mongoid/contextual.rb
+++ b/lib/mongoid/contextual.rb
@@ -35,6 +35,18 @@ def context
       @context ||= create_context
     end
 
+    # Instructs the context to schedule an asynchronous loading of documents
+    # specified by the criteria.
+    #
+    # Note that depending on the context and on the Mongoid configuration,
+    # documents can be loaded synchronously on the caller's thread.
+    #
+    # @return [ Criteria ] Returns self.
+    def load_async
+      context.load_async if context.respond_to?(:load_async)
+      self
+    end
+
     private
 
     # Create the context for the queries to execute. Will be memory for
diff --git a/lib/mongoid/contextual/mongo.rb b/lib/mongoid/contextual/mongo.rb
index 1df5607c5c..6b0e373833 100644
--- a/lib/mongoid/contextual/mongo.rb
+++ b/lib/mongoid/contextual/mongo.rb
@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require "mongoid/contextual/mongo/documents_loader"
 require "mongoid/contextual/atomic"
 require "mongoid/contextual/aggregable/mongo"
 require "mongoid/contextual/command"
@@ -37,6 +38,8 @@ class Mongo
       # @attribute [r] view The Mongo collection view.
       attr_reader :view
 
+      attr_reader :documents_loader
+
       # Get the number of documents matching the query.
       #
       # @example Get the number of matching documents.
@@ -777,6 +780,17 @@ def third_to_last!
         third_to_last || raise_document_not_found_error
       end
 
+      # Schedule a task to load documents for the context.
+      #
+      # Depending on the Mongoid configuration, the scheduled task can be executed
+      # immediately on the caller's thread, or can be scheduled for an
+      # asynchronous execution.
+      #
+      # @api private
+      def load_async
+        @documents_loader ||= DocumentsLoader.new(view, klass, criteria)
+      end
+
       private
 
       # Update the documents for the provided method.
@@ -844,24 +858,29 @@ def inverse_sorting
         Hash[sort.map{|k, v| [k, -1*v]}]
       end
 
-      # Get the documents the context should iterate. This follows 3 rules:
-      #
-      # 1. If the query is cached, and we already have documents loaded, use
-      #   them.
-      # 2. If we are eager loading, then eager load the documents and use
-      #   those.
-      # 3. Use the query.
-      #
-      # @api private
+      # Get the documents the context should iterate.
       #
-      # @example Get the documents for iteration.
-      #   context.documents_for_iteration
+      # If the documents have been already preloaded by `Document::Loader`
+      # instance, they will be used.
       #
       # @return [ Array<Document> | Mongo::Collection::View ] The docs to iterate.
+      #
+      # @api private
       def documents_for_iteration
-        return view unless eager_loadable?
-        docs = view.map{ |doc| Factory.from_db(klass, doc, criteria) }
-        eager_load(docs)
+        if @documents_loader
+          if @documents_loader.started?
+            @documents_loader.value!
+          else
+            @documents_loader.unschedule
+            @documents_loader.execute
+          end
+        else
+          return view unless eager_loadable?
+          docs = view.map do |doc|
+            Factory.from_db(klass, doc, criteria)
+          end
+          eager_load(docs)
+        end
       end
 
       # Yield to the document.
diff --git a/lib/mongoid/contextual/mongo/documents_loader.rb b/lib/mongoid/contextual/mongo/documents_loader.rb
new file mode 100644
index 0000000000..0ed7efea22
--- /dev/null
+++ b/lib/mongoid/contextual/mongo/documents_loader.rb
@@ -0,0 +1,177 @@
+require "mongoid/association/eager_loadable"
+
+module Mongoid
+  module Contextual
+    class Mongo
+      # Loads documents for the provided criteria.
+      #
+      # @api private
+      class DocumentsLoader
+        extend Forwardable
+        include Association::EagerLoadable
+
+        def_delegators :@future, :value!, :value, :wait!, :wait
+
+        # Returns synchronous executor to be used when async_query_executor config option
+        # is set to :immediate. This executor runs all operations on the current
+        # thread, blocking as necessary.
+        #
+        # @return [ Concurrent::ImmediateExecutor ] The executor
+        #   to be used to execute document loading tasks.
+        def self.immediate_executor
+          @@immediate_executor ||= Concurrent::ImmediateExecutor.new
+        end
+
+        # Returns asynchronous executor to be used when async_query_executor config option
+        # is set to :global_thread_pool. This executor runs operations on background threads
+        # using a thread pool.
+        #
+        # @return [ Concurrent::ThreadPoolExecutor ] The executor
+        #   to be used to execute document loading tasks.
+        def self.global_thread_pool_async_query_executor
+          create_pool = Proc.new do |concurrency|
+            Concurrent::ThreadPoolExecutor.new(
+              min_threads: 0,
+              max_threads: concurrency,
+              max_queue: concurrency * 4,
+              fallback_policy: :caller_runs
+            )
+          end
+          concurrency = Mongoid.global_executor_concurrency || 4
+          @@global_thread_pool_async_query_executor ||= create_pool.call(concurrency)
+          if @@global_thread_pool_async_query_executor.max_length != concurrency
+            old_pool = @@global_thread_pool_async_query_executor
+            @@global_thread_pool_async_query_executor = create_pool.call(concurrency)
+            old_pool.shutdown
+          end
+          @@global_thread_pool_async_query_executor
+        end
+
+        # Returns suitable executor according to Mongoid config options.
+        #
+        # @param [ String | Symbol] name The query executor name, can be either
+        #   :immediate or :global_thread_pool. Defaulted to `async_query_executor`
+        #   config option.
+        #
+        # @return [ Concurrent::ImmediateExecutor | Concurrent::ThreadPoolExecutor ] The executor
+        #   to be used to execute document loading tasks.
+        #
+        # @raise [ Errors::InvalidQueryExecutor ] If an unknown name is provided.
+        def self.executor(name = Mongoid.async_query_executor)
+          case name.to_sym
+          when :immediate
+            immediate_executor
+          when :global_thread_pool
+            global_thread_pool_async_query_executor
+          else
+            raise Errors::InvalidQueryExecutor.new(name)
+          end
+        end
+
+        # @return [ Mongoid::Criteria ] Criteria that specifies which documents should
+        #   be loaded. Exposed here because `eager_loadable?` method from
+        #   `Association::EagerLoadable` expects this to be available.
+        attr_accessor :criteria
+
+        # Instantiates the document loader instance and immediately schedules
+        # its execution using the provided executor.
+        #
+        # @param [ Mongo::Collection::View ] view The collection view to get
+        #   records from the database.
+        # @param [ Class ] klass Mongoid model class to instantiate documents.
+        #   All records obtained from the database will be converted to an
+        #   instance of this class, if possible.
+        # @param [ Mongoid::Criteria ] criteria. Criteria that specifies which
+        #   documents should be loaded.
+        # @param [ Concurrent::AbstractExecutorService ] executor. Executor that
+        #   is capable of running `Concurrent::Promises::Future` instances.
+        def initialize(view, klass, criteria, executor: self.class.executor)
+          @view = view
+          @klass = klass
+          @criteria = criteria
+          @mutex = Mutex.new
+          @state = :pending
+          @future = Concurrent::Promises.future_on(executor) do
+            start && execute
+          end
+        end
+
+        # Returns false or true whether the loader is in pending state.
+        #
+        # Pending state means that the loader execution has been scheduled,
+        # but has not been started yet.
+        #
+        # @return [ true | false ] true if the loader is in pending state,
+        #   otherwise false.
+        def pending?
+          @mutex.synchronize do
+            @state == :pending
+          end
+        end
+
+        # Returns false or true whether the loader is in started state.
+        #
+        # Started state means that the loader execution has been started.
+        # Note that the loader stays in this state even after the execution
+        # completed (successfully or failed).
+        #
+        # @return [ true | false ] true if the loader is in started state,
+        #   otherwise false.
+        def started?
+          @mutex.synchronize do
+            @state == :started
+          end
+        end
+
+        # Mark the loader as unscheduled.
+        #
+        # If the loader is marked unscheduled, it will not be executed. The only
+        # option to load the documents is to call `execute` method directly.
+        #
+        # Please note that if execution of a task has been already started,
+        # unscheduling does not have any effect.
+        def unschedule
+          @mutex.synchronize do
+            @state = :cancelled unless @state == :started
+          end
+        end
+
+        # Loads records specified by `@criteria` from the database, and convert
+        # them to Mongoid documents of `@klass` type.
+        #
+        # This method is called by the task (possibly asynchronous) scheduled
+        # when creating an instance of the loader. However, this method can be
+        # called directly, if it is desired to execute loading on the caller
+        # thread immediately.
+        #
+        # Calling this method does not change the state of the loader.
+        #
+        # @return [ Array<Mongoid::Document> ] Array of document loaded from
+        #   the database.
+        def execute
+          documents = @view.map do |doc|
+            Factory.from_db(@klass, doc, @criteria)
+          end
+          eager_load(documents) if eager_loadable?
+          documents
+        end
+
+        private
+
+        # Mark the loader as started if possible.
+        #
+        # @return [ true | false ] Whether the state was changed to :started.
+        def start
+          @mutex.synchronize do
+            if @state == :pending
+              @state = :started
+              true
+            else
+              false
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/errors.rb b/lib/mongoid/errors.rb
index 2dd534ce1a..6d3b32a938 100644
--- a/lib/mongoid/errors.rb
+++ b/lib/mongoid/errors.rb
@@ -10,6 +10,7 @@
 require "mongoid/errors/document_not_found"
 require "mongoid/errors/empty_config_file"
 require "mongoid/errors/in_memory_collation_not_supported"
+require "mongoid/errors/invalid_async_query_executor"
 require "mongoid/errors/invalid_collection"
 require "mongoid/errors/invalid_config_file"
 require "mongoid/errors/invalid_config_option"
@@ -18,6 +19,7 @@
 require "mongoid/errors/invalid_field_option"
 require "mongoid/errors/invalid_field_type"
 require "mongoid/errors/invalid_find"
+require "mongoid/errors/invalid_global_executor_concurrency"
 require "mongoid/errors/invalid_includes"
 require "mongoid/errors/invalid_index"
 require "mongoid/errors/invalid_options"
diff --git a/lib/mongoid/errors/invalid_async_query_executor.rb b/lib/mongoid/errors/invalid_async_query_executor.rb
new file mode 100644
index 0000000000..180e7b7dc3
--- /dev/null
+++ b/lib/mongoid/errors/invalid_async_query_executor.rb
@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+module Mongoid
+  module Errors
+
+    # This error is raised when a bad async query executor option is attempted
+    # to be set.
+    class InvalidQueryExecutor < MongoidError
+
+      # Create the new error.
+      #
+      # @param [ Symbol | String ] executor The attempted async query executor.
+      #
+      # @api private
+      def initialize(executor)
+        super(
+          compose_message(
+            "invalid_async_query_executor",
+            { executor: executor, options: [:immediate, :global_thread_pool] }
+          )
+        )
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/errors/invalid_global_executor_concurrency.rb b/lib/mongoid/errors/invalid_global_executor_concurrency.rb
new file mode 100644
index 0000000000..0cb842e493
--- /dev/null
+++ b/lib/mongoid/errors/invalid_global_executor_concurrency.rb
@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+module Mongoid
+  module Errors
+
+    # This error is raised when a bad global executor concurrency option is attempted
+    # to be set.
+    class InvalidGlobalExecutorConcurrency < MongoidError
+
+      # Create the new error.
+      #
+      # @api private
+      def initialize
+        super(
+          compose_message(
+            "invalid_global_executor_concurrency"
+          )
+        )
+      end
+    end
+  end
+end
diff --git a/lib/mongoid/extensions/time_with_zone.rb b/lib/mongoid/extensions/time_with_zone.rb
index 48739bc141..10c17cc3e5 100644
--- a/lib/mongoid/extensions/time_with_zone.rb
+++ b/lib/mongoid/extensions/time_with_zone.rb
@@ -26,9 +26,10 @@ def mongoize
       end
 
       # This code is copied from Time class extension in bson-ruby gem. It
-      # should be removed from here when added to bson-ruby.
-      # See https://jira.mongodb.org/browse/RUBY-2846.
+      # should be removed from here when the minimum BSON version is 5+.
+      # See https://jira.mongodb.org/browse/MONGOID-5491.
       def _bson_to_i
+        return super if defined?(super)
         # Workaround for JRuby's #to_i rounding negative timestamps up
         # rather than down (https://github.com/jruby/jruby/issues/6104)
         if BSON::Environment.jruby?
diff --git a/mongoid.gemspec b/mongoid.gemspec
index ffbb650483..732de484dd 100644
--- a/mongoid.gemspec
+++ b/mongoid.gemspec
@@ -38,6 +38,7 @@ Gem::Specification.new do |s|
   # See: https://github.com/rails/rails/pull/43951
   s.add_dependency("activemodel", ['>=5.1', '<7.1', '!= 7.0.0'])
   s.add_dependency("mongo", ['>=2.18.0', '<3.0.0'])
+  s.add_dependency("concurrent-ruby", ['>= 1.0.5', '<1.2'])
 
   # The ruby2_keywords gem is recommended for handling argument delegation issues,
   # especially if support for 2.6 or prior is required.
diff --git a/spec/mongoid/association/referenced/has_and_belongs_to_many/buildable_spec.rb b/spec/mongoid/association/referenced/has_and_belongs_to_many/buildable_spec.rb
index cf89504bdd..9a8df0be7e 100644
--- a/spec/mongoid/association/referenced/has_and_belongs_to_many/buildable_spec.rb
+++ b/spec/mongoid/association/referenced/has_and_belongs_to_many/buildable_spec.rb
@@ -111,7 +111,7 @@
         end
 
         let(:criteria) do
-          Preference.all_of("_id" => { "$in" => [] })
+          Preference.none
         end
 
         it "a criteria object" do
diff --git a/spec/mongoid/config_spec.rb b/spec/mongoid/config_spec.rb
index 4ac5d4bca0..d402d1775e 100644
--- a/spec/mongoid/config_spec.rb
+++ b/spec/mongoid/config_spec.rb
@@ -224,6 +224,79 @@
     end
   end
 
+  context 'async_query_executor option' do
+    let(:option) { :async_query_executor }
+
+    before do
+      Mongoid::Config.reset
+      Mongoid.configure do |config|
+        config.load_configuration(conf)
+      end
+    end
+
+    context "when it is not set in the config" do
+
+      let(:conf) { CONFIG }
+
+      it "it is set to its default" do
+        expect(Mongoid.send(option)).to eq(:immediate)
+      end
+    end
+
+    context 'when the value is :immediate' do
+
+      let(:conf) do
+        CONFIG.merge(options: { option => :immediate })
+      end
+
+      it "is set to false" do
+        expect(Mongoid.send(option)).to be(:immediate)
+      end
+    end
+
+    context 'when the value is :global_thread_pool' do
+
+      let(:conf) do
+        CONFIG.merge(options: { option => :global_thread_pool })
+      end
+
+      it "is set to false" do
+        expect(Mongoid.send(option)).to be(:global_thread_pool)
+      end
+    end
+  end
+
+  context 'global_executor_concurrency option' do
+    let(:option) { :global_executor_concurrency }
+
+    before do
+      Mongoid::Config.reset
+      Mongoid.configure do |config|
+        config.load_configuration(conf)
+      end
+    end
+
+    context "when it is not set in the config" do
+
+      let(:conf) { CONFIG }
+
+      it "it is set to its default" do
+        expect(Mongoid.send(option)).to eq(nil)
+      end
+    end
+
+    context 'when the value is set to a number' do
+
+      let(:conf) do
+        CONFIG.merge(options: { option => 5 })
+      end
+
+      it "is set to the number" do
+        expect(Mongoid.send(option)).to be(5)
+      end
+    end
+  end
+
   shared_examples "a config option" do
 
     before do
@@ -652,6 +725,17 @@ def self.logger
         }.to raise_error(Mongoid::Errors::InvalidConfigOption)
       end
     end
+
+    context 'when invalid global_executor_concurrency option provided' do
+      it "raises an error" do
+        expect do
+          described_class.options = {
+            async_query_executor: :immediate,
+            global_executor_concurrency: 5
+          }
+        end.to raise_error(Mongoid::Errors::InvalidGlobalExecutorConcurrency)
+      end
+    end
   end
 
   describe "#clients=" do
diff --git a/spec/mongoid/contextual/mongo/documents_loader_spec.rb b/spec/mongoid/contextual/mongo/documents_loader_spec.rb
new file mode 100644
index 0000000000..936186c54f
--- /dev/null
+++ b/spec/mongoid/contextual/mongo/documents_loader_spec.rb
@@ -0,0 +1,187 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+describe Mongoid::Contextual::Mongo::DocumentsLoader do
+  # https://jira.mongodb.org/browse/MONGOID-5505
+  require_mri
+
+  let(:view) do
+    double('view').tap do |view|
+      allow(view).to receive(:map)
+    end
+  end
+
+  let(:klass) do
+    double
+  end
+
+  let(:criteria) do
+    double('criteria').tap do |criteria|
+      allow(criteria).to receive(:inclusions).and_return([])
+    end
+  end
+
+  let(:executor) do
+    described_class.executor
+  end
+
+  let(:subject) do
+    described_class.new(view, klass, criteria, executor: executor)
+  end
+
+  context 'state management' do
+    let(:executor) do
+      # Such executor will never execute a task, so it guarantees that
+      # our task will stay in its initial state.
+      Concurrent::ThreadPoolExecutor.new(
+        min_threads: 0,
+        max_threads: 0,
+      )
+    end
+
+    describe '#initialize' do
+      it 'initializes in pending state' do
+        expect(subject.pending?).to be_truthy
+        expect(subject.started?).to be_falsey
+      end
+    end
+
+    describe '#unschedule' do
+      it 'changes state' do
+        subject.unschedule
+        expect(subject.pending?).to be_falsey
+        expect(subject.started?).to be_falsey
+      end
+    end
+
+    describe '#execute' do
+      it 'does not change state' do
+        prev_started = subject.started?
+        prev_pending = subject.pending?
+        subject.execute
+        expect(subject.started?).to eq(prev_started)
+        expect(subject.pending?).to eq(prev_pending)
+      end
+    end
+
+    context 'when the task is completed' do
+      let(:executor) do
+        Concurrent::ImmediateExecutor.new
+      end
+
+      it 'changes the state to started' do
+        subject.wait!
+        expect(subject.started?).to be_truthy
+        expect(subject.pending?).to be_falsey
+      end
+    end
+  end
+
+  context 'loading documents' do
+    let(:view) do
+      klass.collection.find(criteria.selector, session: criteria.send(:_session))
+    end
+
+    let(:criteria) do
+      Band.where(name: "Depeche Mode")
+    end
+
+    let(:klass) do
+      Band
+    end
+
+    let!(:band) do
+      Band.create!(name: 'Depeche Mode')
+    end
+
+    context 'asynchronously' do
+      it 'loads documents' do
+        subject.wait!
+        expect(subject.value).to eq([band])
+      end
+    end
+
+    context 'synchronously' do
+      let(:executor) do
+        # Such executor will never execute a task, so it guarantees that
+        # our task will stay in its initial state.
+        Concurrent::ThreadPoolExecutor.new(
+          min_threads: 0,
+          max_threads: 0,
+        )
+      end
+
+      it 'loads documents' do
+        expect(subject.execute).to eq([band])
+      end
+    end
+  end
+
+  describe '.executor' do
+    context 'when immediate executor requested' do
+      it 'returns immediate executor' do
+        expect(
+          described_class.executor(:immediate)
+        ).to eq(described_class.immediate_executor)
+      end
+    end
+
+    context 'when global thread pool executor requested' do
+      it 'returns global thread pool executor' do
+        expect(
+          described_class.executor(:global_thread_pool)
+        ).to eq(described_class.global_thread_pool_async_query_executor)
+      end
+    end
+
+    context 'when an unknown executor requested' do
+      it 'raises an error' do
+        expect do
+          described_class.executor(:i_am_an_invalid_option)
+        end.to raise_error(Mongoid::Errors::InvalidQueryExecutor)
+      end
+    end
+  end
+
+  describe ".global_thread_pool_async_query_executor" do
+    before(:each) do
+      described_class.class_variable_set(:@@global_thread_pool_async_query_executor, nil)
+    end
+
+    after(:each) do
+      described_class.class_variable_set(:@@global_thread_pool_async_query_executor, nil)
+    end
+
+    context 'when global_executor_concurrency option is set' do
+      config_override :global_executor_concurrency, 50
+
+      it 'returns an executor' do
+        executor = described_class.global_thread_pool_async_query_executor
+        expect(executor).not_to be_nil
+        expect(executor.max_length).to eq( 50 )
+      end
+    end
+
+    context 'when global_executor_concurrency option is not set' do
+      it 'returns an executor' do
+        executor = described_class.global_thread_pool_async_query_executor
+        expect(executor).not_to be_nil
+        expect(executor.max_length).to eq( 4 )
+      end
+    end
+
+    context 'when global_executor_concurrency option changes' do
+      config_override :global_executor_concurrency, 50
+
+      it 'creates new executor' do
+        first_executor = described_class.global_thread_pool_async_query_executor
+        Mongoid.global_executor_concurrency = 100
+        second_executor = described_class.global_thread_pool_async_query_executor
+
+        expect(first_executor).not_to eq(second_executor)
+        expect(second_executor.max_length).to eq( 100 )
+      end
+    end
+  end
+end
diff --git a/spec/mongoid/contextual/mongo_spec.rb b/spec/mongoid/contextual/mongo_spec.rb
index 01e24e4ea3..b346c88101 100644
--- a/spec/mongoid/contextual/mongo_spec.rb
+++ b/spec/mongoid/contextual/mongo_spec.rb
@@ -4473,4 +4473,70 @@
       end
     end
   end
+
+  describe '#load_async' do
+    let!(:band) do
+      Band.create!(name: "Depeche Mode")
+    end
+
+    let(:criteria) do
+      Band.where(name: "Depeche Mode")
+    end
+
+    let(:context) do
+      described_class.new(criteria)
+    end
+
+    context 'with global thread pool async query executor' do
+      config_override :async_query_executor, :global_thread_pool
+
+      it 'preloads the documents' do
+        context.load_async
+        context.documents_loader.wait
+
+        expect(context.view).not_to receive(:map)
+        expect(context.to_a).to eq([band])
+      end
+
+      it 're-raises exception during preload' do
+        expect_any_instance_of(Mongoid::Contextual::Mongo::DocumentsLoader)
+          .to receive(:execute)
+          .at_least(:once)
+          .and_raise(Mongo::Error::OperationFailure)
+
+        context.load_async
+        context.documents_loader.wait
+
+        expect do
+          context.to_a
+        end.to raise_error(Mongo::Error::OperationFailure)
+      end
+    end
+
+    context 'with immediate thread pool async query executor' do
+      config_override :async_query_executor, :immediate
+
+      it 'preloads the documents' do
+        context.load_async
+        context.documents_loader.wait
+
+        expect(context.view).not_to receive(:map)
+        expect(context.to_a).to eq([band])
+      end
+
+      it 're-raises exception during preload' do
+        expect_any_instance_of(Mongoid::Contextual::Mongo::DocumentsLoader)
+          .to receive(:execute)
+          .at_least(:once)
+          .and_raise(Mongo::Error::OperationFailure)
+
+        context.load_async
+        context.documents_loader.wait
+
+        expect do
+          context.to_a
+        end.to raise_error(Mongo::Error::OperationFailure)
+      end
+    end
+  end
 end
diff --git a/spec/mongoid/scopable_spec.rb b/spec/mongoid/scopable_spec.rb
index 132194e8d5..060fa0d960 100644
--- a/spec/mongoid/scopable_spec.rb
+++ b/spec/mongoid/scopable_spec.rb
@@ -121,6 +121,76 @@
         expect(selector).to eq({'active' => true})
       end
     end
+
+    context "when the default scope is dotted" do
+
+      let(:criteria) do
+        Band.where('tags.foo' => 'bar')
+      end
+
+      before do
+        Band.default_scope ->{ criteria }
+      end
+
+      after do
+        Band.default_scoping = nil
+      end
+
+      let!(:band) do
+        Band.create!
+      end
+
+      it "adds the scope as a dotted key attribute" do
+        expect(band.attributes['tags.foo']).to eq('bar')
+      end
+
+      it "adds the default scope to the class" do
+        expect(Band.default_scoping.call).to eq(criteria)
+      end
+
+      it "flags as being default scoped" do
+        expect(Band).to be_default_scoping
+      end
+
+      it "does not find the correct document" do
+        expect(Band.count).to eq(0)
+      end
+    end
+
+    context "when the default scope is dotted with a query" do
+
+      let(:criteria) do
+        Band.where('tags.foo' => {'$eq' => 'bar'})
+      end
+
+      before do
+        Band.default_scope ->{ criteria }
+      end
+
+      after do
+        Band.default_scoping = nil
+      end
+
+      let!(:band) do
+        Band.create!('tags' => { 'foo' => 'bar' })
+      end
+
+      it "does not add the scope as a dotted key attribute" do
+        expect(band.attributes).to_not have_key('tags.foo')
+      end
+
+      it "adds the default scope to the class" do
+        expect(Band.default_scoping.call).to eq(criteria)
+      end
+
+      it "flags as being default scoped" do
+        expect(Band).to be_default_scoping
+      end
+
+      it "finds the correct document" do
+        expect(Band.where.first).to eq(band)
+      end
+    end
   end
 
   describe ".default_scopable?" do
diff --git a/spec/support/models/band.rb b/spec/support/models/band.rb
index e142a0dc84..6723ffcbff 100644
--- a/spec/support/models/band.rb
+++ b/spec/support/models/band.rb
@@ -21,6 +21,7 @@ class Band
   field :founded, type: Date
   field :deleted, type: Boolean
   field :mojo, type: Object
+  field :tags, type: Hash
   field :fans
 
   embeds_many :records, cascade_callbacks: true