diff --git a/guides/dataloader/built_in_sources.md b/guides/dataloader/built_in_sources.md
new file mode 100644
index 0000000000..515644aea7
--- /dev/null
+++ b/guides/dataloader/built_in_sources.md
@@ -0,0 +1,16 @@
+---
+layout: guide
+doc_stub: false
+search: true
+section: Dataloader
+title: Built-in source
+desc: Default Dataloader sources in GraphQL-Ruby
+index: 2
+---
+
+Although you'll probably need some {% internal_link "custom sources", "/dataloader/custom_sources" %} before long, GraphQL-Ruby ships with a few basic sources to get you started and serve as examples. Follow the links below to see the API docs for each source:
+
+- {{ "GraphQL::Dataloader::ActiveRecord" | api_doc }}
+- {{ "GraphQL::Dataloader::ActiveRecordAssociation" | api_doc }}
+- {{ "GraphQL::Dataloader::Http" | api_doc }}
+- {{ "GraphQL::Dataloader::Redis" | api_doc }}
diff --git a/guides/dataloader/custom_sources.md b/guides/dataloader/custom_sources.md
new file mode 100644
index 0000000000..3220e8bd95
--- /dev/null
+++ b/guides/dataloader/custom_sources.md
@@ -0,0 +1,119 @@
+---
+layout: guide
+doc_stub: false
+search: true
+section: Dataloader
+title: Custom sources
+desc: Writing a custom Dataloader source for GraphQL-Ruby
+index: 3
+---
+
+To write a custom dataloader source, you have to consider a few points:
+
+- Batch keys: these inputs tell the dataloader how work can be batched
+- Fetch parameters: these inputs are accumulated into batches, and dispatched all at once
+- Executing the service call: How to take inputs and group them into an external call
+- Handling the results: mapping the results of the external call back to the fetch parameters
+
+Additionally, custom sources can perform their service calls in [background threads](#background-threads).
+
+For this example, we'll imagine writing a Dataloader source for a non-ActiveRecord SQL backend.
+
+## Batch Keys, Fetch Parameters
+
+`GraphQL::Dataloader` assumes that external sources have two kinds of parameters:
+
+- __Batch keys__ are parameters which _distinguish_ batches from one another; calls with different batch keys are resolved in different batches.
+- __Fetch parameters__ are parameters which _merge_ into batches; calls with the same batch keys but different fetch parameters are merged in the same batch.
+
+Looking at SQL:
+
+- tables are _batch keys_: objects from different tables will be resolved in different batches.
+- IDs are _fetch parameters_: objects with different IDs may be fetched in the same batch (given that they're on the same table).
+
+With this in mind, our source's public API will look like this:
+
+```ruby
+# To request a user by ID:
+SQLDatabase.load("users", user_id)
+#                ^^^^^^^           <- Batch key (table name)
+#                         ^^^^^^^  <- Fetch parameter (id)
+```
+
+With an API like that, the source could be used for general purpose ID lookups:
+
+```ruby
+SQLDatabase.load("products", product_id_1)   #  <
+SQLDatabase.load("products", product_id_2)   #  < These two will be resolved in the batch
+
+SQLDatabase.load("reviews", review_id)       #  < This will be resolved in a different batch
+```
+
+{{ "GraphQL::Dataloader::Source.load" | api_doc }} assumes that the final argument is a _fetch parameter_ and that all other arguments (if there are any) are batch keys. So, our Source class won't need to modify that method.
+
+However, we'll want to capture the table name for each batch, and we'll use `#intialize` for that:
+
+```ruby
+class SQLDatabase < GraphQL::Dataloader::Source
+  def initialize(table_name)
+    # Next, we'll use `@table_name` to prepare a SQL query, see below
+    @table_name = table_name
+  end
+end
+```
+
+Each time GraphQL-Ruby encounters a new batch key, it initializes a Source for that key. Then, while the query is running, that Source will be reused for all calls to that batch key. (GraphQL-Ruby clears the source cache between mutations.)
+
+## Executing the Service Call and Handling the Results
+
+Source classes must implement `#perform(fetch_parameters)` to call the data source, retrieve values, and fulfill each fetch parameter. `#perform` is called by GraphQL internals when it has determined that no further execution is possible without resolving a batch load operation.
+
+In our case, we'll use the batch key (table name) and fetch parameters (IDs) to construct a SQL query. Then, we'll dispatch the query to get results. Finally, we'll get the object for each ID and fulfill the ID.
+
+```ruby
+class SQLDatabase < GraphQL::Dataloader::Source
+  def initialize(table_name)
+    @table_name = table_name
+  end
+
+  def perform(ids)
+    if ids.any? { |id| !id.is_a?(Numeric) }
+      raise ArgumentError, "Invalid IDs: #{ids}"
+    end
+
+    if !@table_name.match?(/\A[a-z_]+\Z/)
+      raise ArgumentError, "Invalid table name: #{@table_name}"
+    end
+
+    # Prepare a query and send it to the database
+    query = "SELECT * FROM #{@table_name} WHERE id IN(#{ids.join(",")})"
+    results = DatabaseConnection.execute(query)
+
+    # Then, for each of the given `ids`, find the matching result (or `nil`)
+    # and call `fulfill(id, result)` to tell GraphQL-Ruby what object to use for that ID.
+    ids.each do |id|
+      result = results.find { |r| r.id == id }
+      fulfill(id, result)
+    end
+  end
+end
+```
+
+During `fulfill`, GraphQL-Ruby caches the `id => result` pair. Any subsequent loads to that ID will return the previously-fetched result.
+
+## Background Threads
+
+You can tell GraphQL-Ruby to call `#perform` in a background thread by including {{ "GraphQL::Dataloader::Source::BackgroundThreaded" | api_doc }}. For example:
+
+```ruby
+class SQLDatabase < GraphQL::Dataloader::Source
+  # This class's `perform` method will be called in the background
+  include GraphQL::Dataloader::Source::BackgroundThreaded
+end
+```
+
+Under the hood, GraphQL-Ruby uses [`Concurrent::Promises::Future`](https://ruby-concurrency.github.io/concurrent-ruby/1.1.7/Concurrent/Promises/Future.html) from [concurrent-ruby](https://github.com/ruby-concurrency/concurrent-ruby/). Add to your Gemfile:
+
+```ruby
+gem "concurrent-ruby", require: "concurrent"
+```
diff --git a/guides/dataloader/overview.md b/guides/dataloader/overview.md
new file mode 100644
index 0000000000..6d7e7009ba
--- /dev/null
+++ b/guides/dataloader/overview.md
@@ -0,0 +1,58 @@
+---
+layout: guide
+doc_stub: false
+search: true
+section: Dataloader
+title: Overview
+desc: Data loading in GraphQL
+index: 0
+redirect_from:
+  - /schema/lazy_execution
+---
+
+Because GraphQL queries are very dynamic, GraphQL systems require a different approach to fetching data into your application. Here, we'll discuss the problem and solution at a conceptual level. Later, the {% internal_link "Using Dataloader", "/dataloader/usage" %} and {% internal_link "Custom Sources", "/dataloader/custom_sources" %} guides provide concrete implementation advice.
+
+## Dynamic Data Requirements
+
+When your application renders a hardcoded HTML template or JSON payload, you can customize your SQL query for minimum overhead and maximum performance. But, in GraphQL, the response is highly dependent on the incoming query. When clients are sending custom queries, you can't hand-tune database queries!
+
+For example, imagine this incoming GraphQL query:
+
+```ruby
+films(first: 10) {
+  director { name }
+}
+```
+
+If the `director` field is implemented with a Rails `belongs_to` association, it will be an N+1 situation by default. As each `Film`'s fields are resolved, they will each dispatch a SQL query:
+
+```SQL
+SELECT * FROM directors WHERE id = 1;
+SELECT * FROM directors WHERE id = 2;
+SELECT * FROM directors WHERE id = 3;
+...
+```
+
+This is inefficient because we make _many_ round-trips to the database. So, how can we improve our GraphQL system to use that more-efficient query?
+
+(Although this example uses SQL, the same issue applies to any external service that your application might fetch data from, for example: Redis, Memcached, REST APIs, GraphQL APIs, search engines, RPC servers.)
+
+## Batching External Service Calls
+
+The solution is to dispatch service calls in _batches_. As a GraphQL query runs, you can gather up information, then finally dispatch a call. In the example above, we could _batch_ those SQL queries into a single query:
+
+```SQL
+SELECT * FROM directors WHERE id IN(1,2,3,...);
+```
+
+This technique was demonstrated in [graphql/dataloader](https://github.com/graphql/dataloader) and implemented in Ruby by [shopify/graphql-batch](https://github.com/shopify/graphql-batch) and [exaspark/batch-loader](https://github.com/exAspArk/batch-loader/). Now, GraphQL-Ruby has a built-in implementation, {{ "GraphQL::Dataloader" | api_doc }}.
+
+## GraphQL::Dataloader
+
+{{ "GraphQL::Dataloader" | api_doc }} is an implementation of batch loading for GraphQL-Ruby. It consists of several components:
+
+- {{ "GraphQL::Dataloader" | api_doc }} instances, which manage a cache of sources during query execution
+- {{ "GraphQL::Dataloader::Source" | api_doc }}, a base class for batching calls to data layers and caching the results
+- {{ "GrpahQL::Execution::Lazy" | api_doc }}, a Promise-like object which can be chained with `.then { ... }` or zipped with `GraphQL::Execution::Lazy.all(...)`.
+
+Check out the {% internal_link "Usage guide", "dataloader/usage" %} to get started with it.
diff --git a/guides/dataloader/usage.md b/guides/dataloader/usage.md
new file mode 100644
index 0000000000..26304ba781
--- /dev/null
+++ b/guides/dataloader/usage.md
@@ -0,0 +1,87 @@
+---
+layout: guide
+doc_stub: false
+search: true
+section: Dataloader
+title: Usage
+desc: Getting started with GraphQL::Dataloader
+index: 1
+---
+
+To add {{ "GraphQL::Dataloader" | api_doc }} to your schema, attach it with `use`:
+
+```ruby
+class MySchema < GraphQL::Schema
+  # ...
+  use GraphQL::Dataloader
+end
+```
+
+## Batch-loading data
+
+With {{ "GraphQL::Dataloader" | api_doc }} in your schema, you're ready to start batch loading data. For example:
+
+```ruby
+class Types::Post < Types::BaseObject
+  field :author, Types::Author, null: true, description: "The author who wrote this post"
+
+  def author
+    # Look up this Post's author by its `belongs_to` association
+    GraphQL::Dataloader::ActiveRecordAssociation.load(:author, object)
+  end
+end
+```
+
+Or, load data from a URL:
+
+```ruby
+class Types::User < Types::BaseObject
+  field :github_repos_count, Integer, null: true,
+    description: "The number of repos this person has on GitHub"
+
+  def github_repos_count
+    # Fetch some JSON, then return one of the values from it.
+    GraphQL::Dataloader::Http.load("https://api.github.com/users/#{object.github_login}").then do |data|
+      data["public_repos"]
+    end
+  end
+end
+```
+
+{{ "GraphQL::Dataloader::ActiveRecordAssociation" | api_doc }} and {{ "GraphQL::Dataloader::Http" | api_doc }} are _source classes_ which fields can use to request data. Under the hood, GraphQL will defer the _actual_ data fetching as long as possible, so that batches can be gathered up and sent together.
+
+For a full list of built-in sources, see the {% internal_link "Built-in sources guide", "/dataloader/built_in_sources" %}.
+
+To write custom sources, see the {% internal_link "Custom sources guide", "/dataloader/custom_sources" %}.
+
+## Node IDs
+
+With {{ "GraphQL::Dataloader" | api_doc }}, you can batch-load objects inside `MySchema.object_from_id`:
+
+```ruby
+class MySchema < GraphQL::Schema
+  def self.object_from_id(id, ctx)
+    # TODO update graphql-ruby's defaults to support this
+    model_class, model_id = MyIdScheme.decode(id)
+    GraphQL::Dataloader::ActiveRecord.load(model_class, model_id)
+  end
+end
+```
+
+This way, even `loads:` IDs will be batch loaded, for example:
+
+```ruby
+class Types::Query < Types::BaseObject
+  field :post, Types::Post, null: true,
+    description: "Look up a post by ID" do
+      argument :id, ID, required: true, loads: Types::Post, as: :post
+    end
+  end
+
+  def post(post:)
+    post
+  end
+end
+```
+
+To learn about available sources, see the {% internal_link "built-in sources guide", "/dataloader/built_in_sources" %}. Or, check out the {% internal_link "custom sources guide", "/dataloader/custom_sources" %} to get started with your own sources.
diff --git a/guides/schema/definition.md b/guides/schema/definition.md
index 7c36f0710d..c811a09a50 100644
--- a/guides/schema/definition.md
+++ b/guides/schema/definition.md
@@ -138,14 +138,6 @@ class MySchema < GraphQL::Schema
 end
 ```
 
-__`lazy_resolve`__ registers classes with {% internal_link "lazy execution", "/schema/lazy_execution" %}:
-
-```ruby
-class MySchema < GraphQL::Schema
-  lazy_resolve Promise, :sync
-end
-```
-
 __`type_error`__ handles type errors at runtime, read more in the {% internal_link "Invariants guide", "/errors/type_errors" %}.
 
 ```ruby
diff --git a/guides/schema/lazy_execution.md b/guides/schema/lazy_execution.md
deleted file mode 100644
index aa958d26ed..0000000000
--- a/guides/schema/lazy_execution.md
+++ /dev/null
@@ -1,95 +0,0 @@
----
-layout: guide
-doc_stub: false
-search: true
-title: Lazy Execution
-section: Schema
-desc: Resolve functions can return "unfinished" results that are deferred for batch resolution.
-index: 4
----
-
-With lazy execution, you can optimize access to external services (such as databases) by making batched calls. Building a lazy loader has three steps:
-
-- Define a lazy-loading class with _one_ method for loading & returning a value
-- Connect it to your schema with {{ "GraphQL::Schema#lazy_resolve" | api_doc }}
-- In `resolve` functions, return instances of the lazy-loading class
-
-## Example: Batched Find
-
-Here's a way to find many objects by ID using one database call, preventing N+1 queries.
-
-1. Lazy-loading class which finds models by ID.
-
-```ruby
-class LazyFindPerson
-  def initialize(query_ctx, person_id)
-    @person_id = person_id
-    # Initialize the loading state for this query,
-    # or get the previously-initiated state
-    @lazy_state = query_ctx[:lazy_find_person] ||= {
-      pending_ids: Set.new,
-      loaded_ids: {},
-    }
-    # Register this ID to be loaded later:
-    @lazy_state[:pending_ids] << person_id
-  end
-
-  # Return the loaded record, hitting the database if needed
-  def person
-    # Check if the record was already loaded:
-    loaded_record = @lazy_state[:loaded_ids][@person_id]
-    if loaded_record
-      # The pending IDs were already loaded,
-      # so return the result of that previous load
-      loaded_record
-    else
-      # The record hasn't been loaded yet, so
-      # hit the database with all pending IDs
-      pending_ids = @lazy_state[:pending_ids].to_a
-      people = Person.where(id: pending_ids)
-      people.each { |person| @lazy_state[:loaded_ids][person.id] = person }
-      @lazy_state[:pending_ids].clear
-      # Now, get the matching person from the loaded result:
-      @lazy_state[:loaded_ids][@person_id]
-    end
-  end
-```
-
-2. Connect the lazy resolve method
-
-```ruby
-class MySchema < GraphQL::Schema
-  # ...
-  lazy_resolve(LazyFindPerson, :person)
-end
-```
-
-3. Return lazy objects from `resolve`
-
-```ruby
-field :author, PersonType, null: true
-
-def author
-  LazyFindPerson.new(context, object.author_id)
-end
-```
-
-Now, calls to `author` will use batched database access. For example, this query:
-
-```graphql
-{
-  p1: post(id: 1) { author { name } }
-  p2: post(id: 2) { author { name } }
-  p3: post(id: 3) { author { name } }
-}
-```
-
-Will only make one query to load the `author` values.
-
-## Gems for batching
-
-The example above is simple and has some shortcomings. Consider the following gems for a robust solution to batched resolution:
-
-* [`graphql-batch`](https://github.com/shopify/graphql-batch) provides a powerful, flexible toolkit for lazy resolution with GraphQL.
-* [`dataloader`](https://github.com/sheerun/dataloader) is more general promise-based utility for batching queries within the same thread.
-* [`batch-loader`](https://github.com/exAspArk/batch-loader) works with any Ruby code including GraphQL, no extra dependencies or primitives.
diff --git a/lib/graphql.rb b/lib/graphql.rb
index 8c671226bf..0afbaffdf0 100644
--- a/lib/graphql.rb
+++ b/lib/graphql.rb
@@ -148,3 +148,5 @@ def match?(pattern)
 require "graphql/unauthorized_error"
 require "graphql/unauthorized_field_error"
 require "graphql/load_application_object_failed_error"
+require "graphql/pagination"
+require "graphql/dataloader"
diff --git a/lib/graphql/dataloader.rb b/lib/graphql/dataloader.rb
new file mode 100644
index 0000000000..ac5c8edc5f
--- /dev/null
+++ b/lib/graphql/dataloader.rb
@@ -0,0 +1,125 @@
+# frozen_string_literal: true
+require "graphql/dataloader/source"
+require "graphql/dataloader/active_record"
+require "graphql/dataloader/active_record_association"
+require "graphql/dataloader/http"
+require "graphql/dataloader/instrumentation"
+require "graphql/dataloader/load_error"
+require "graphql/dataloader/mutation_field_extension"
+require "graphql/dataloader/no_dataloader_error"
+require "graphql/dataloader/redis"
+
+module GraphQL
+  class Dataloader
+    def self.use(schema)
+      instrumenter = Dataloader::Instrumentation.new
+      schema.instrument(:multiplex, instrumenter)
+      # TODO this won't work if the mutation is hooked up after this
+      schema.mutation && schema.mutation.fields.each do |name, field|
+        field.extension(MutationFieldExtension)
+      end
+    end
+
+    class << self
+      # @return [Dataloader, nil] The dataloader instance caching loaders for the current thread
+      def current
+        Thread.current[:graphql_dataloader]
+      end
+
+      def current=(dataloader)
+        Thread.current[:graphql_dataloader] = dataloader
+      end
+
+      # Call the given block using the provided dataloader
+      # @param dataloader [Dataloader] A new one is created if one isn't given.
+      def load(dataloader = Dataloader.new(nil))
+        result = begin
+          begin_dataloading(dataloader)
+          yield
+        ensure
+          end_dataloading
+        end
+
+        GraphQL::Execution::Lazy.sync(result)
+      end
+
+      def begin_dataloading(dataloader = Dataloader.new(nil))
+        self.current ||= dataloader
+        increment_level
+      end
+
+      def end_dataloading
+        decrement_level
+        if level < 1
+          self.current = nil
+        end
+      end
+
+      private
+
+      def level
+        @level || 0
+      end
+
+      def increment_level
+        @level ||= 0
+        @level += 1
+      end
+
+      def decrement_level
+        @level ||= 0
+        @level -= 1
+      end
+    end
+
+    def initialize(multiplex)
+      @multiplex = multiplex
+
+      @sources = Concurrent::Map.new do |h, source_class|
+        h[source_class] = Concurrent::Map.new do |h2, source_key|
+          # TODO maybe add `cache_key` API like graphql-batch has
+          h2[source_key] = source_class.new(*source_key)
+        end
+      end
+
+      @async_source_queue = []
+    end
+
+    # @param source_class [Class]
+    # @param source_key [Object] A cache key for instances of `source_class`
+    # @return [Dataloader::Source] an instance of `source_class` for `key`, cached for the duration of the multiplex
+    def source_for(source_class, source_key)
+      @sources[source_class][source_key]
+    end
+
+    def current_query
+      @multiplex && @multiplex.context[:current_query]
+    end
+
+    # Clear the cached loaders of this dataloader (eg, after running a mutation).
+    # @return void
+    def clear
+      @sources.clear
+      nil
+    end
+
+    # Register this source for background processing.
+    # @param source [Dataloader::Source]
+    # @return void
+    # @api private
+    def enqueue_async_source(source)
+      if !@async_source_queue.include?(source)
+        @async_source_queue << source
+      end
+    end
+
+    # Call `.wait` on each pending background loader, and clear the queue.
+    # @return void
+    # @api private
+    def process_async_source_queue
+      queue = @async_source_queue
+      @async_source_queue = []
+      queue.each(&:wait)
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/active_record.rb b/lib/graphql/dataloader/active_record.rb
new file mode 100644
index 0000000000..2e3a5f24b9
--- /dev/null
+++ b/lib/graphql/dataloader/active_record.rb
@@ -0,0 +1,39 @@
+# frozen_string_literal.rb
+
+module GraphQL
+  class Dataloader
+    # @example Find a record by ID
+    #   GraphQL::Dataloader::ActiveRecord.load(Post, id)
+    #
+    # @example Find several records by their IDs
+    #   GraphQL::Dataloader::ActiveRecord.load_all(Post, [id1, id2, id3])
+    class ActiveRecord < Dataloader::Source
+      def initialize(model, column: model.primary_key)
+        @model = model
+        @column = column
+        @column_type = model.type_for_attribute(@column)
+      end
+
+      # Override this to make sure that the values always match type (eg, turn `"1"` into `1`)
+      def load(column_value)
+        casted_value = if @column_type.respond_to?(:type_cast)
+          @column_type.type_cast(column_value)
+        elsif @column_type.respond_to?(:type_cast_from_user)
+          @column_type.type_cast_from_user(column_value)
+        else
+          @column_type.cast(column_value)
+        end
+
+        super(casted_value)
+      end
+
+      def perform(column_values)
+        records = @model.where(@column => column_values)
+        column_values.each do |v|
+          record = records.find { |r| r.public_send(@column) == v }
+          fulfill(v, record)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/active_record_association.rb b/lib/graphql/dataloader/active_record_association.rb
new file mode 100644
index 0000000000..74defe5858
--- /dev/null
+++ b/lib/graphql/dataloader/active_record_association.rb
@@ -0,0 +1,44 @@
+# frozen_string_literal.rb
+
+module GraphQL
+  class Dataloader
+    # @example Load belongs-to associations in a batch
+    #   ActiveRecordAssociation.load(Post, :author, post_1)
+    #   ActiveRecordAssociation.load(Post, :author, post_2)
+    class ActiveRecordAssociation < Dataloader::Source
+      def initialize(model, association_name)
+        @model = model
+        @association_name = association_name
+      end
+
+      def self.load(association_name, record)
+        super(record.class, association_name, record)
+      end
+
+      def load(record)
+        # return early if this association is already loaded
+        if record.association(@association_name).loaded?
+          GraphQL::Execution::Lazy.new { record.public_send(@association_name) }
+        else
+          super
+        end
+      end
+
+      def perform(records)
+        if ::ActiveRecord::Associations::Preloader.method_defined?(:call)
+          # After Rails 6.2, Preloader's API changes to `new(**kwargs).call`
+          ::ActiveRecord::Associations::Preloader
+            .new(records: records, associations: @association_name, scope: nil)
+            .call
+        else
+          ::ActiveRecord::Associations::Preloader
+            .new
+            .preload(records, @association_name)
+        end
+        records.each { |record|
+          fulfill(record, record.public_send(@association_name))
+        }
+      end
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/http.rb b/lib/graphql/dataloader/http.rb
new file mode 100644
index 0000000000..29d2954540
--- /dev/null
+++ b/lib/graphql/dataloader/http.rb
@@ -0,0 +1,28 @@
+# frozen_string_literal.rb
+require "net/http"
+
+module GraphQL
+  class Dataloader
+    # This source performs an HTTP GET for each given URL, then returns the parsed JSON body.
+    #
+    # In reality, this source should check `response.content_type` and
+    # handle non-success responses in some way, but it doesn't.
+    #
+    # For your own application, you'd probably want to batch calls to external APIs in semantically useful ways,
+    # where IDs or call parameters are grouped, then merged in to an API call to a known URL.
+    class Http < Dataloader::Source
+      def perform(urls)
+        urls.each do |url|
+          uri = URI(url)
+          response = Net::HTTP.get_response(uri)
+          parsed_body = if response.body.empty?
+            nil
+          else
+            JSON.parse(response.body)
+          end
+          fulfill(url, parsed_body)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/instrumentation.rb b/lib/graphql/dataloader/instrumentation.rb
new file mode 100644
index 0000000000..93a5c4bfb1
--- /dev/null
+++ b/lib/graphql/dataloader/instrumentation.rb
@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+module GraphQL
+  class Dataloader
+    class Instrumentation
+      def before_multiplex(multiplex)
+        dataloader = Dataloader.new(multiplex)
+        Dataloader.begin_dataloading(dataloader)
+      end
+
+      def after_multiplex(_m)
+        Dataloader.end_dataloading
+      end
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/load_error.rb b/lib/graphql/dataloader/load_error.rb
new file mode 100644
index 0000000000..c869c881b2
--- /dev/null
+++ b/lib/graphql/dataloader/load_error.rb
@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+module GraphQL
+  class Dataloader
+    class LoadError < GraphQL::Error
+      # @return [Array<Integer, String>] The runtime GraphQL path where the failed load was requested
+      attr_accessor :graphql_path
+
+      attr_writer :message
+
+      def message
+        @message || super
+      end
+
+      attr_writer :cause
+
+      def cause
+        @cause || super
+      end
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/mutation_field_extension.rb b/lib/graphql/dataloader/mutation_field_extension.rb
new file mode 100644
index 0000000000..2c44c66ab9
--- /dev/null
+++ b/lib/graphql/dataloader/mutation_field_extension.rb
@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+module GraphQL
+  class Dataloader
+    class MutationFieldExtension < GraphQL::Schema::FieldExtension
+      def resolve(object:, arguments:, context:, **_rest)
+        Dataloader.current.clear
+        begin
+          return_value = yield(object, arguments)
+          GraphQL::Execution::Lazy.sync(return_value)
+        ensure
+          Dataloader.current.clear
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/no_dataloader_error.rb b/lib/graphql/dataloader/no_dataloader_error.rb
new file mode 100644
index 0000000000..437b8d696f
--- /dev/null
+++ b/lib/graphql/dataloader/no_dataloader_error.rb
@@ -0,0 +1,8 @@
+# frozen_string_literal: true
+
+module GraphQL
+  class Dataloader
+    class NoDataloaderError < GraphQL::Error
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/redis.rb b/lib/graphql/dataloader/redis.rb
new file mode 100644
index 0000000000..0b59283df5
--- /dev/null
+++ b/lib/graphql/dataloader/redis.rb
@@ -0,0 +1,37 @@
+# frozen_string_literal.rb
+
+module GraphQL
+  class Dataloader
+    # This source uses Redis pipelining to execute a bunch of commands.
+    #
+    # In practice, an application-specific source would be more appropriate, because you
+    # could choose between commands like GET and MGET, HGETALL and HMGET, etc.
+    #
+    # But this source is here as an example of what's possible.
+    #
+    # @example Getting values from a redis connection.
+    #
+    #   GraphQL::Dataloader::Redis.load($redis, [:get, "some-key"])
+    #   GraphQL::Dataloader::Redis.load($redis, [:hgetall, "some-hash-key"])
+    #   GraphQL::Dataloader::Redis.load($redis, [:smembers, "some-set-key"])
+    #
+    class Redis < Dataloader::Source
+      def initialize(redis_connection)
+        @redis = redis_connection
+      end
+
+      def perform(commands)
+        results = @redis.pipelined do
+          commands.map do |(command, *args)|
+            @redis.public_send(command, *args)
+          end
+        end
+
+        commands.each_with_index do |command, idx|
+          result = results[idx]
+          fulfill(command, result)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/source.rb b/lib/graphql/dataloader/source.rb
new file mode 100644
index 0000000000..a8fab4b6c4
--- /dev/null
+++ b/lib/graphql/dataloader/source.rb
@@ -0,0 +1,96 @@
+# frozen_string_literal: true
+require "graphql/dataloader/source/background_threaded"
+
+module GraphQL
+  class Dataloader
+    class Source
+      # @return [GraphQL::Execution::Lazy]
+      def self.load(*key, value)
+        self.for(*key).load(value)
+      end
+
+      # @return [Source] A cached instance of this class for the given {key_parts}
+      def self.for(*key_parts)
+        dl = Dataloader.current
+        if !dl
+          raise Dataloader::NoDataloaderError, "Can't initialize a Source without a Dataloader, use `Dataloader.load { ... }` or add `use GraphQL::Dataloader` to your schema"
+        end
+        dl.source_for(self, key_parts)
+      end
+
+      # @return [GraphQL::Execution::Lazy]
+      def self.load_all(*key_parts, values)
+        pending_loads = values.map { |value| load(*key_parts, value) }
+        Execution::Lazy.all(pending_loads)
+      end
+
+      # @see .load for a cache-friendly way to load objects during a query
+      def load(key)
+        pending_loads.compute_if_absent(key) { make_lazy(key) }
+      end
+
+      # Called by {Execution::Lazy}s that are waiting for this loader
+      # @api private
+      def wait
+        # loads might be added in the meantime, but they won't be included in this list.
+        keys_to_load = @load_queue
+        @load_queue = nil
+        perform_with_error_handling(keys_to_load)
+      end
+
+      # Mark {key} as having loaded {value}
+      # @return void
+      def fulfill(key, value)
+        pending_loads[key].fulfill(value)
+        nil
+      end
+
+      # @return [Boolean] true if `key` was loaded
+      def fulfilled?(key)
+        (lazy = pending_loads[key]) && lazy.resolved?
+      end
+
+      # This method should take `keys` and load a value for each one, then call `fulfill(k, value || nil)` to mark the load as successful
+      # @param keys [Array<Object>] Whatever values have been passed to {#load} since this source's last perform call
+      # @return void
+      def perform(keys)
+        raise RequiredImplementationMissingError, "`#{self.class}#perform(keys)` should call `fulfill(key, loaded_value)` for each of `keys`"
+      end
+
+      private
+
+      def fulfilled_value_for(key)
+        # TODO raise if not loaded?
+        (lazy = pending_loads[key]) && lazy.value
+      end
+
+      def pending_loads
+        @pending_loads ||= Concurrent::Map.new
+      end
+
+      def perform_with_error_handling(keys_to_load)
+        perform(keys_to_load)
+        nil
+      rescue GraphQL::ExecutionError
+        # Allow client-facing errors to keep propagating
+        raise
+      rescue StandardError => cause
+        message = "Error from #{self.class}#perform(#{keys_to_load.map(&:inspect).join(", ")})\n\n#{cause.class}:\n#{cause.message.inspect}"
+        load_err = GraphQL::Dataloader::LoadError.new(message)
+        load_err.set_backtrace(cause.backtrace)
+        load_err.cause = cause
+
+        keys_to_load.each do |key|
+          fulfill(key, load_err)
+        end
+        raise load_err
+      end
+
+      def make_lazy(key)
+        @load_queue ||= []
+        @load_queue << key
+        Execution::Lazy.new(self)
+      end
+    end
+  end
+end
diff --git a/lib/graphql/dataloader/source/background_threaded.rb b/lib/graphql/dataloader/source/background_threaded.rb
new file mode 100644
index 0000000000..3cc7c73412
--- /dev/null
+++ b/lib/graphql/dataloader/source/background_threaded.rb
@@ -0,0 +1,55 @@
+# frozen_string_literal: true
+
+module GraphQL
+  class Dataloader
+    class Source
+      # Include this module to make Source subclasses run their {#perform} methods inside `Concurrent::Promises.future { ... }`
+      module BackgroundThreaded
+        # Assert that concurrent-ruby is present
+        def self.included(_child_class)
+          if !defined?(Concurrent)
+            raise "concurrent-ruby is required to use #{self}, add `gem 'concurrent-ruby', require: 'concurrent'` to your Gemfile and `bundle install`"
+          end
+        end
+
+        private
+
+        # This is called when populating the promise cache.
+        # In this case, also register the source for async processing.
+        # (It might have already been registered by another `key`, the dataloader will ignore it in that case.)
+        def make_lazy(key)
+          lazy = super
+          Dataloader.current.enqueue_async_source(self)
+          lazy
+        end
+
+        # Like the superclass method, but:
+        #
+        # - Wrap the call to `super` inside a `Concurrent::Promises::Future`
+        # - In the meantime, `fulfill(...)` each key with a lazy that will wait for the future
+        #
+        # Interestingly, that future will `fulfill(...)` each key with a finished value, so only the first
+        # of the Lazies will actually be called. (Since the others will be replaced.)
+        def perform_with_error_handling(keys_to_load)
+          this_dl = Dataloader.current
+          future = Concurrent::Promises.delay do
+            Dataloader.load(this_dl) do
+              super(keys_to_load)
+            end
+          end
+
+          keys_to_load.each do |key|
+            lazy = GraphQL::Execution::Lazy.new do
+              future.value # force waiting for it to be finished
+              fulfilled_value_for(key)
+            end
+            fulfill(key, lazy)
+          end
+          # Actually kick off the future:
+          future.touch
+          nil
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/execution/interpreter.rb b/lib/graphql/execution/interpreter.rb
index 63c8c408e6..9d775b65ce 100644
--- a/lib/graphql/execution/interpreter.rb
+++ b/lib/graphql/execution/interpreter.rb
@@ -94,8 +94,11 @@ def sync_lazies(query: nil, multiplex: nil)
         end
         final_values.compact!
         tracer.trace("execute_query_lazy", {multiplex: multiplex, query: query}) do
+          multiplex.context[:current_query] = query
           Interpreter::Resolve.resolve_all(final_values)
         end
+        multiplex.context[:current_query] = nil
+
         queries.each do |query|
           runtime = query.context.namespace(:interpreter)[:runtime]
           if runtime
diff --git a/lib/graphql/execution/interpreter/resolve.rb b/lib/graphql/execution/interpreter/resolve.rb
index ad94d55e57..f35d076a1b 100644
--- a/lib/graphql/execution/interpreter/resolve.rb
+++ b/lib/graphql/execution/interpreter/resolve.rb
@@ -25,8 +25,10 @@ def self.resolve_all(results)
         # or return {Hash}/{Array} if the query should be continued.
         #
         # @param results [Array]
-        # @return [Array] Same size, filled with finished values
+        # @return [Array] The next round of lazies to resolve
         def self.resolve(results)
+          # First, kick off any sources that will resolve in background threads
+          Dataloader.current && Dataloader.current.process_async_source_queue
           next_results = []
 
           # Work through the queue until it's empty
diff --git a/lib/graphql/execution/lazy.rb b/lib/graphql/execution/lazy.rb
index 2de4d10a4b..13a7a3b390 100644
--- a/lib/graphql/execution/lazy.rb
+++ b/lib/graphql/execution/lazy.rb
@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 require "graphql/execution/lazy/lazy_method_map"
 require "graphql/execution/lazy/resolve"
+require "graphql/execution/lazy/group"
 
 module GraphQL
   module Execution
@@ -8,10 +9,9 @@ module Execution
     #
     # Calling `#value` will trigger calculation & return the "lazy" value.
     #
-    # This is an itty-bitty promise-like object, with key differences:
+    # This is a promise-like object, with key differences:
     # - It has only two states, not-resolved and resolved
     # - It has no error-catching functionality
-    # @api private
     class Lazy
       # Traverse `val`, lazily resolving any values along the way
       # @param val [Object] A data structure containing mixed plain values and `Lazy` instances
@@ -20,54 +20,170 @@ def self.resolve(val)
         Resolve.resolve(val)
       end
 
-      attr_reader :path, :field
+      # If `maybe_lazy` is a Lazy, sync it recursively. (Never returns another Lazy)
+      def self.sync(maybe_lazy)
+        while maybe_lazy.is_a?(Lazy)
+          maybe_lazy = maybe_lazy.value
+        end
+
+        maybe_lazy
+      end
+
+      # @param maybe_lazies [Array<Lazy, Object>]
+      # @return [Lazy] A single lazy wrapping any number of maybe-lazy objects
+      def self.all(maybe_lazies)
+        group = Group.new(maybe_lazies)
+        group.lazy
+      end
+
+      class OnlyBlockSource
+        def initialize(block, promise)
+          @block = block
+          @promise = promise
+        end
+
+        def wait
+          value = @block.call
+          @promise.fulfill(value)
+        end
+      end
+
+      # @return [Array<String, Integer>] The runtime path where this lazy was created
+      attr_reader :path
 
-      # Create a {Lazy} which will get its inner value by calling the block
+      # @return [GraphQL::Schema::Field] The field that was executing when this lazy was created
+      attr_reader :field
+
+      # Create a {Lazy} which will get its inner value from `source,` and/or by calling the block
+      # @param source [<#wait>] Some object that this Lazy depends on, if there is one
       # @param path [Array<String, Integer>]
       # @param field [GraphQL::Schema::Field]
-      # @param get_value_func [Proc] a block to get the inner value (later)
-      def initialize(path: nil, field: nil, &get_value_func)
-        @get_value_func = get_value_func
+      # @param then_block [Proc] a block to get the inner value (later)
+      def initialize(source = nil, path: nil, field: nil, caller_offset: 0, &then_block)
+        @caller = caller(2 + caller_offset, 1).first
+        if source.nil?
+          # This lazy is just a deferred block
+          @source = OnlyBlockSource.new(then_block, self)
+          @then_block = nil
+        else
+          @source = source
+          @then_block = then_block
+        end
         @resolved = false
+        @value = nil
+        @pending_lazies = nil
         @path = path
         @field = field
       end
 
       # @return [Object] The wrapped value, calling the lazy block if necessary
+      # @raise [StandardError] if this lazy was {#fulfill}ed with an error
       def value
+        wait
+        if @value.is_a?(StandardError)
+          raise @value
+        else
+          @value
+        end
+      end
+
+      def sync
+        self.class.sync(self)
+      end
+
+      # resolve this lazy's dependencies as long as one can be found
+      # @return [void]
+      def wait
         if !@resolved
-          @resolved = true
-          @value = begin
-            v = @get_value_func.call
-            if v.is_a?(Lazy)
-              v = v.value
+          while (current_source = @source)
+            current_source.wait
+            # Only care if these are the same object,
+            # which shows that the lazy didn't start
+            # waiting on something else
+            if current_source.equal?(@source)
+              break
             end
-            v
-          rescue GraphQL::ExecutionError => err
-            err
           end
         end
+      rescue GraphQL::Dataloader::LoadError => err
+        raise tag_error(err)
+      end
 
-        if @value.is_a?(StandardError)
-          raise @value
+      # @return [Boolean] true if this value has received a finished value, by a direct call to {#fulfill} or by {#wait}, which might trigger a call to {#fulfill}
+      def resolved?
+        @resolved
+      end
+
+      # Set this Lazy's resolved value, if it hasn't already been resolved.
+      #
+      # @param value [Object] If this is a `StandardError`, it will be raised by {#value}
+      # @param call_then [Boolean] When `true`, this Lazy's `@then_block` will be called with `value` before passing it along
+      # @return [void]
+      def fulfill(value, call_then: false)
+        if @resolved
+          return
+        end
+
+        if call_then && @then_block
+          value = @then_block.call(value)
+        end
+
+        if value.is_a?(Lazy)
+          if value.resolved?
+            fulfill(value.value)
+          else
+            @source = value
+            value.subscribe(self, call_then: false)
+          end
         else
-          @value
+          @source = nil
+          @resolved = true
+          @value = value
+          if @pending_lazies
+            non_error = !value.is_a?(StandardError)
+            lazies = @pending_lazies
+            @pending_lazies = nil
+            lazies.each { |lazy, call_then|
+              lazy.fulfill(value, call_then: non_error && call_then)
+            }
+          end
         end
       end
 
-      # @return [Lazy] A {Lazy} whose value depends on another {Lazy}, plus any transformations in `block`
-      def then
-        self.class.new {
-          yield(value)
-        }
+      # @return [Lazy] A {Lazy} whose value depends on `self`, plus any transformations in `block`
+      def then(&block)
+        new_lazy = self.class.new(self, &block)
+        subscribe(new_lazy, call_then: true)
+        new_lazy
+      end
+
+      def inspect
+        "#<#{self.class.name}##{object_id} from #{@caller.inspect} / #{@field.respond_to?(:path) ? "#{@field.path} " : ""}#{@path ? "#{@path} " : ""}@source=#<#{@source.class}> @resolved=#{@resolved} @value=#{@value.inspect}>"
       end
 
-      # @param lazies [Array<Object>] Maybe-lazy objects
-      # @return [Lazy] A lazy which will sync all of `lazies`
-      def self.all(lazies)
-        self.new {
-          lazies.map { |l| l.is_a?(Lazy) ? l.value : l }
-        }
+      protected
+
+      # Add `other_lazy` to this Lazy's dependencies. It will be `.fulfill(...)`ed with the value of this lazy, eventually.
+      def subscribe(other_lazy, call_then:)
+        @pending_lazies ||= []
+        @pending_lazies.push([other_lazy, call_then])
+      end
+
+      private
+
+      # Copy `err` and update the copy with Lazy-specfic details.
+      # (The error probably came from a Source, which has any number of pending Lazies.)
+      # @param err [Dataloader::LoadError]
+      # @return [Dataloader::LoadError] An updated copy
+      def tag_error(err)
+        local_err = err.dup
+        query = Dataloader.current.current_query
+        if query
+          local_err.graphql_path = query.context[:current_path]
+          op_name = query.selected_operation_name || query.selected_operation.operation_type || "query"
+          local_err.message = err.message.sub(")\n\n", ") at #{op_name}.#{local_err.graphql_path.join(".")}\n\n")
+        end
+        local_err
       end
 
       # This can be used for fields which _had no_ lazy results
diff --git a/lib/graphql/execution/lazy/group.rb b/lib/graphql/execution/lazy/group.rb
new file mode 100644
index 0000000000..e8a098241e
--- /dev/null
+++ b/lib/graphql/execution/lazy/group.rb
@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+module GraphQL
+  module Execution
+    class Lazy
+      # @see Lazy.all, this is an implementation detail of that method
+      class Group
+        attr_reader :lazy
+
+        def initialize(maybe_lazies)
+          @lazy = Lazy.new(self)
+          @maybe_lazies = maybe_lazies
+          @waited = false
+        end
+
+        def wait
+          if !@waited
+            @waited = true
+            any_lazies = false
+            results = @maybe_lazies.map { |maybe_lazy|
+              if maybe_lazy.respond_to?(:wait)
+                maybe_lazy.wait
+                res = maybe_lazy.value
+                any_lazies ||= res.is_a?(Lazy)
+                res
+              else
+                maybe_lazy
+              end
+            }
+
+            if any_lazies
+              results = Lazy.all(results)
+            end
+
+            lazy.fulfill(results)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/execution/lazy/resolve.rb b/lib/graphql/execution/lazy/resolve.rb
index c19415f117..c00a9ef70a 100644
--- a/lib/graphql/execution/lazy/resolve.rb
+++ b/lib/graphql/execution/lazy/resolve.rb
@@ -31,13 +31,13 @@ def self.resolve(value)
 
         def self.resolve_in_place(value)
           acc = each_lazy(NullAccumulator, value)
-
           if acc.empty?
             Lazy::NullResult
           else
             Lazy.new {
               acc.each_with_index { |ctx, idx|
-                acc[idx] = ctx.value.value
+                v = Lazy.sync(ctx.value.value)
+                acc[idx] = v
               }
               resolve_in_place(acc)
             }
diff --git a/lib/graphql/execution/multiplex.rb b/lib/graphql/execution/multiplex.rb
index 87ac53ca84..93506e65a3 100644
--- a/lib/graphql/execution/multiplex.rb
+++ b/lib/graphql/execution/multiplex.rb
@@ -103,6 +103,7 @@ def run_as_multiplex(multiplex)
         # @param query [GraphQL::Query]
         # @return [Hash] The initial result (may not be finished if there are lazy values)
         def begin_query(query, multiplex)
+          multiplex.context[:current_query] = query
           operation = query.selected_operation
           if operation.nil? || !query.valid? || query.context.errors.any?
             NO_OPERATION
@@ -115,12 +116,15 @@ def begin_query(query, multiplex)
               NO_OPERATION
             end
           end
+        ensure
+          multiplex.context[:current_query] = nil
         end
 
         # @param data_result [Hash] The result for the "data" key, if any
         # @param query [GraphQL::Query] The query which was run
         # @return [Hash] final result of this query, including all values and errors
         def finish_query(data_result, query, multiplex)
+          multiplex.context[:current_query] = query
           # Assign the result so that it can be accessed in instrumentation
           query.result_values = if data_result.equal?(NO_OPERATION)
             if !query.valid? || query.context.errors.any?
@@ -140,6 +144,8 @@ def finish_query(data_result, query, multiplex)
 
             result
           end
+        ensure
+          multiplex.context[:current_query] = nil
         end
 
         # use the old `query_execution_strategy` etc to run this query
diff --git a/lib/graphql/schema.rb b/lib/graphql/schema.rb
index 501b70ac2f..adeb9d8d20 100644
--- a/lib/graphql/schema.rb
+++ b/lib/graphql/schema.rb
@@ -105,7 +105,7 @@ module LazyHandlingMethods
       # @api private
       def after_lazy(value, &block)
         if lazy?(value)
-          GraphQL::Execution::Lazy.new do
+          GraphQL::Execution::Lazy.new(caller_offset: 1) do
             result = sync_lazy(value)
             # The returned result might also be lazy, so check it, too
             after_lazy(result, &block)
@@ -281,7 +281,7 @@ def initialize
       @parse_error_proc = DefaultParseError
       @instrumenters = Hash.new { |h, k| h[k] = [] }
       @lazy_methods = GraphQL::Execution::Lazy::LazyMethodMap.new
-      @lazy_methods.set(GraphQL::Execution::Lazy, :value)
+      @lazy_methods.set(GraphQL::Execution::Lazy, :sync)
       @cursor_encoder = Base64Encoder
       # For schema instances, default to legacy runtime modules
       @analysis_engine = GraphQL::Analysis
@@ -1710,7 +1710,7 @@ def lazy_methods
             @lazy_methods = inherited_map.dup
           else
             @lazy_methods = GraphQL::Execution::Lazy::LazyMethodMap.new
-            @lazy_methods.set(GraphQL::Execution::Lazy, :value)
+            @lazy_methods.set(GraphQL::Execution::Lazy, :sync)
           end
         end
         @lazy_methods
diff --git a/spec/graphql/dataloader/active_record_association_spec.rb b/spec/graphql/dataloader/active_record_association_spec.rb
new file mode 100644
index 0000000000..783d6a089f
--- /dev/null
+++ b/spec/graphql/dataloader/active_record_association_spec.rb
@@ -0,0 +1,129 @@
+# frozen_string_literal: true
+require "spec_helper"
+
+# Rails 3 doesn't have type_for_attribute
+if testing_rails? && ActiveRecord::Base.respond_to?(:type_for_attribute)
+  describe GraphQL::Dataloader::ActiveRecordAssociation do
+    class Artist < ActiveRecord::Base
+      has_many :albums
+    end
+
+    class Album < ActiveRecord::Base
+      belongs_to :artist
+    end
+
+    the_shins = Artist.create!(name: "The Shins")
+    the_shins.albums.create!(name: "Oh, Inverted World")
+    the_shins.albums.create!(name: "Chutes Too Narrow")
+    the_shins.albums.create!(name: "Wincing the Night Away")
+
+    mt_joy = Artist.create!(name: "Mt. Joy")
+    mt_joy.albums.create!(name: "Mt. Joy")
+    mt_joy.albums.create!(name: "Rearrange Us")
+
+    the_extraordinaires = Artist.create(name: "The Extraordinaires")
+    the_extraordinaires.albums.create!(name: "Ribbons of War")
+    the_extraordinaires.albums.create!(name: "Short Stories")
+    the_extraordinaires.albums.create!(name: "Electric and Benevolent")
+    the_extraordinaires.albums.create!(name: "Home Sweet Home")
+
+    class DataloaderActiveRecordAssociationSchema < GraphQL::Schema
+      class Query < GraphQL::Schema::Object
+        field :artist_album_count, Integer, null: true do
+          argument :album_name, String, required: true
+        end
+
+        def artist_album_count(album_name:)
+          GraphQL::Dataloader::ActiveRecord.for(Album, column: "name").load(album_name).then do |album|
+            # IRL This could be done better using `album.artist_id`, but this is a nice way to test the belongs-to association
+            album && GraphQL::Dataloader::ActiveRecordAssociation.load(:artist, album).then do |artist|
+              artist && artist.albums.count
+            end
+          end
+        end
+
+        field :artist_name, String, null: true do
+          argument :album_name, String, required: true
+        end
+
+        def artist_name(album_name:)
+          GraphQL::Dataloader::ActiveRecord.for(Album, column: "name").load(album_name).then do |album|
+            album && GraphQL::Dataloader::ActiveRecordAssociation.load(:artist, album).then(&:name)
+          end
+        end
+      end
+
+      query(Query)
+      use GraphQL::Dataloader
+    end
+
+    def exec_query(*args, **kwargs)
+      DataloaderActiveRecordAssociationSchema.execute(*args, **kwargs)
+    end
+
+    it "batches calls for belongs-tos" do
+      res = nil
+      log = []
+      callback = ->(_name, _start, _end, _digest, query, *rest) { log << [query[:sql], query[:type_casted_binds]] }
+
+      ActiveSupport::Notifications.subscribed(callback, "sql.active_record") do
+        res = exec_query <<-GRAPHQL
+        {
+          ext1: artistAlbumCount(albumName: "Short Stories")
+          shins1: artistName(albumName: "Wincing the Night Away")
+          mtJoy1: artistName(albumName: "Rearrange Us")
+          shins2: artistName(albumName: "Chutes Too Narrow")
+          miss1: artistName(albumName: "Tom's Story")
+          shins3: artistAlbumCount(albumName: "Oh, Inverted World")
+        }
+        GRAPHQL
+      end
+
+      expected_data = {
+        "ext1" => 4,
+        "shins1" => "The Shins",
+        "mtJoy1" => "Mt. Joy",
+        "shins2" => "The Shins",
+        "miss1" => nil,
+        "shins3" => 3,
+      }
+
+      assert_equal(expected_data, res["data"])
+      expected_log = if Rails::VERSION::STRING < "5"
+        # Rails 4
+        [
+          ["SELECT \"albums\".* FROM \"albums\" WHERE \"albums\".\"name\" IN ('Short Stories', 'Wincing the Night Away', 'Rearrange Us', 'Chutes Too Narrow', 'Tom''s Story', 'Oh, Inverted World')", nil],
+          ["SELECT \"artists\".* FROM \"artists\" WHERE \"artists\".\"id\" IN (3, 1, 2)",nil],
+          ["SELECT COUNT(*) FROM \"albums\" WHERE \"albums\".\"artist_id\" = ?", nil],
+          ["SELECT COUNT(*) FROM \"albums\" WHERE \"albums\".\"artist_id\" = ?", nil],
+        ]
+      elsif Rails::VERSION::STRING < "6"
+        # Rails 5
+        [
+          [
+            "SELECT \"albums\".* FROM \"albums\" WHERE \"albums\".\"name\" IN ($1, $2, $3, $4, $5, $6)",
+            ["Short Stories", "Wincing the Night Away", "Rearrange Us", "Chutes Too Narrow", "Tom's Story", "Oh, Inverted World"]
+          ],
+          ["SELECT \"artists\".* FROM \"artists\" WHERE \"artists\".\"id\" IN ($1, $2, $3)", [3, 1, 2]],
+          ["SELECT COUNT(*) FROM \"albums\" WHERE \"albums\".\"artist_id\" = $1", [3]],
+          ["SELECT COUNT(*) FROM \"albums\" WHERE \"albums\".\"artist_id\" = $1", [1]],
+        ]
+      else
+        # Rails 6 +
+        [
+          [
+            "SELECT \"albums\".* FROM \"albums\" WHERE \"albums\".\"name\" IN (?, ?, ?, ?, ?, ?)",
+            ["Short Stories", "Wincing the Night Away", "Rearrange Us", "Chutes Too Narrow", "Tom's Story", "Oh, Inverted World"]
+          ],
+          ["SELECT \"artists\".* FROM \"artists\" WHERE \"artists\".\"id\" IN (?, ?, ?)", [3, 1, 2]],
+          ["SELECT COUNT(*) FROM \"albums\" WHERE \"albums\".\"artist_id\" = ?", [3]],
+          ["SELECT COUNT(*) FROM \"albums\" WHERE \"albums\".\"artist_id\" = ?", [1]],
+        ]
+      end
+
+      if expected_log
+        assert_equal expected_log, log, "It has the expected queries on Rails #{Rails::VERSION::STRING}"
+      end
+    end
+  end
+end
diff --git a/spec/graphql/dataloader/active_record_spec.rb b/spec/graphql/dataloader/active_record_spec.rb
new file mode 100644
index 0000000000..e8c0c6e88e
--- /dev/null
+++ b/spec/graphql/dataloader/active_record_spec.rb
@@ -0,0 +1,116 @@
+# frozen_string_literal: true
+require "spec_helper"
+
+# Rails 3 doesn't have type_for_attribute
+if testing_rails? && ActiveRecord::Base.respond_to?(:type_for_attribute)
+  describe GraphQL::Dataloader::ActiveRecord do
+    class HtmlColor < ActiveRecord::Base
+    end
+
+    HtmlColor.create!(name: "Bisque", hex: 0xFFE4C4)
+    HtmlColor.create!(name: "Thistle", hex: 0xD8BFD8)
+    HtmlColor.create!(name: "Gainsboro", hex: 0xDCDCDC)
+
+    class DataloaderActiveRecordSchema < GraphQL::Schema
+      class Query < GraphQL::Schema::Object
+        field :color_by_hex, String, null: true do
+          argument :hex, String, required: true
+        end
+
+        def color_by_hex(hex:)
+          hex_int = hex.to_i(16)
+          GraphQL::Dataloader::ActiveRecord.for(HtmlColor, column: "hex").load(hex_int).then { |c| c && c.name }
+        end
+
+        field :color_by_id_int, String, null: true do
+          argument :id, Integer, required: true
+        end
+
+        def color_by_id_int(id:)
+          GraphQL::Dataloader::ActiveRecord.load(HtmlColor, id).then { |c| c && c.name }
+        end
+
+        field :color_by_id_str, String, null: true do
+          argument :id, ID, required: true
+        end
+
+        def color_by_id_str(id:)
+          GraphQL::Dataloader::ActiveRecord.load(HtmlColor, id).then { |c| c && c.name }
+        end
+      end
+
+      query(Query)
+      use GraphQL::Dataloader
+    end
+
+    def exec_query(*args, **kwargs)
+      DataloaderActiveRecordSchema.execute(*args, **kwargs)
+    end
+
+    it "calls Model.where with columns and values" do
+      res = nil
+      log = []
+      callback = ->(_name, _start, _end, _digest, query, *rest) { log << [query[:sql], query[:type_casted_binds]] }
+
+      ActiveSupport::Notifications.subscribed(callback, "sql.active_record") do
+        res = exec_query <<-GRAPHQL
+        {
+          bisque1: colorByHex(hex: "FFE4C4")
+          bisque2: colorByIdInt(id: 1)
+          bisque3: colorByIdStr(id: "1")
+          thistle: colorByIdInt(id: 2)
+          gainsboro: colorByIdStr(id: "3")
+          missing: colorByIdInt(id: 99)
+        }
+        GRAPHQL
+      end
+
+      expected_data = {
+        "bisque1" => "Bisque",
+        "bisque2" => "Bisque",
+        "bisque3" => "Bisque",
+        "thistle" => "Thistle",
+        "gainsboro" => "Gainsboro",
+        "missing" => nil
+      }
+
+      assert_equal(expected_data, res["data"])
+
+      expected_log = if Rails::VERSION::STRING < "5"
+        # Rails 4
+        [
+          ["SELECT \"html_colors\".* FROM \"html_colors\" WHERE \"html_colors\".\"hex\" = 16770244", nil],
+          ["SELECT \"html_colors\".* FROM \"html_colors\" WHERE \"html_colors\".\"id\" IN (1, 2, 3, 99)", nil],
+        ]
+      elsif Rails::VERSION::STRING < "6"
+        # Rails 5
+        [
+          [
+            "SELECT \"html_colors\".* FROM \"html_colors\" WHERE \"html_colors\".\"hex\" = $1",
+            [16770244]
+          ],
+          [
+            "SELECT \"html_colors\".* FROM \"html_colors\" WHERE \"html_colors\".\"id\" IN ($1, $2, $3, $4)",
+            [1, 2, 3, 99]
+          ],
+        ]
+      else
+        # Rails 6+
+        [
+          [
+            "SELECT \"html_colors\".* FROM \"html_colors\" WHERE \"html_colors\".\"hex\" = ?",
+            [16770244]
+          ],
+          [
+            "SELECT \"html_colors\".* FROM \"html_colors\" WHERE \"html_colors\".\"id\" IN (?, ?, ?, ?)",
+            [1, 2, 3, 99]
+          ],
+        ]
+      end
+
+      if expected_log
+        assert_equal expected_log, log
+      end
+    end
+  end
+end
diff --git a/spec/graphql/dataloader/batch_compat_spec.rb b/spec/graphql/dataloader/batch_compat_spec.rb
new file mode 100644
index 0000000000..0c6dd4068d
--- /dev/null
+++ b/spec/graphql/dataloader/batch_compat_spec.rb
@@ -0,0 +1,694 @@
+# frozen_string_literal: true
+#
+# GraphQL::Dataloader is basically ripped off from Shopify/graphql-batch,
+# so make sure that it can do everything that graphql-batch can do.
+#
+# Adapted from https://github.com/Shopify/graphql-batch/blob/master/test/graphql_test.rb
+require 'spec_helper'
+
+class GraphQLDataloaderBatchCompatTest < Minitest::Test
+  class QueryNotifier
+    class << self
+      attr_accessor :subscriber
+
+      def call(query)
+        subscriber && subscriber.call(query)
+      end
+    end
+  end
+
+  module ModelClassMethods
+    attr_accessor :fixtures, :has_manys
+
+    def model_name
+      name.split("::").last
+    end
+
+    def first(count)
+      QueryNotifier.call("#{model_name}?limit=#{count}")
+      fixtures.values.first(count).map(&:dup)
+    end
+
+    def find(ids)
+      ids = Array(ids)
+      QueryNotifier.call("#{model_name}/#{ids.join(',')}")
+      ids.map{ |id| fixtures[id] }.compact.map(&:dup)
+    end
+
+    def preload_association(owners, association)
+      association_reflection = reflect_on_association(association)
+      foreign_key = association_reflection[:foreign_key]
+      scope = association_reflection[:scope]
+      rows = association_reflection[:model].fixtures.values
+      owner_ids = owners.map(&:id).to_set
+
+      QueryNotifier.call("#{model_name}/#{owners.map(&:id).join(',')}/#{association}")
+      records = rows.select{ |row|
+        owner_ids.include?(row.public_send(foreign_key)) && scope.call(row)
+      }
+
+      records_by_key = records.group_by(&foreign_key)
+      owners.each do |owner|
+        owner.public_send("#{association}=", records_by_key[owner.id] || [])
+      end
+      nil
+    end
+
+    def has_many(association_name, model:, foreign_key:, scope: ->(row){ true })
+      self.has_manys ||= {}
+      has_manys[association_name] = { model: model, foreign_key: foreign_key, scope: scope }
+      attr_accessor(association_name)
+    end
+
+    def reflect_on_association(association)
+      has_manys.fetch(association)
+    end
+  end
+
+  Image = Struct.new(:id, :owner_type, :owner_id, :filename) do
+    extend ModelClassMethods
+  end
+
+  ProductVariant = Struct.new(:id, :product_id, :title) do
+    extend ModelClassMethods
+    has_many :images, model: Image, foreign_key: :owner_id, scope: ->(row) { row.owner_type == 'ProductVariant' }
+  end
+
+  Product = Struct.new(:id, :title, :image_id) do
+    extend ModelClassMethods
+    has_many :variants, model: ProductVariant, foreign_key: :product_id
+  end
+
+  Product.fixtures = [
+    Product.new(1, "Shirt", 1),
+    Product.new(2, "Pants", 2),
+    Product.new(3, "Sweater", 3),
+  ].each_with_object({}){ |p, h| h[p.id] = p }
+
+  ProductVariant.fixtures = [
+    ProductVariant.new(1, 1, "Red"),
+    ProductVariant.new(2, 1, "Blue"),
+    ProductVariant.new(4, 2, "Small"),
+    ProductVariant.new(5, 2, "Medium"),
+    ProductVariant.new(6, 2, "Large"),
+    ProductVariant.new(7, 3, "Default"),
+  ].each_with_object({}){ |p, h| h[p.id] = p }
+
+  Image.fixtures = [
+    Image.new(1, 'Product', 1, "shirt.jpg"),
+    Image.new(2, 'Product', 2, "pants.jpg"),
+    Image.new(3, 'Product', 3, "sweater.jpg"),
+    Image.new(4, 'ProductVariant', 1, "red-shirt.jpg"),
+    Image.new(5, 'ProductVariant', 2, "blue-shirt.jpg"),
+    Image.new(6, 'ProductVariant', 3, "small-pants.jpg"),
+  ].each_with_object({}){ |p, h| h[p.id] = p }
+
+  class RecordLoader < GraphQL::Dataloader::Source
+    def initialize(model)
+      @model = model
+    end
+
+    def load(id)
+      super(Integer(id))
+    end
+
+    def perform(ids)
+      @model.find(ids).each { |record| fulfill(record.id, record) }
+      ids.each { |id| fulfill(id, nil) unless fulfilled?(id) }
+    end
+  end
+
+  class AssociationLoader < GraphQL::Dataloader::Source
+    def initialize(model, association)
+      @model = model
+      @association = association
+    end
+
+    def perform(owners)
+      @model.preload_association(owners, @association)
+      owners.each { |owner| fulfill(owner, owner.public_send(@association)) }
+    end
+  end
+
+  class CounterLoader < GraphQL::Dataloader::Source
+    def cache_key(counter_array)
+      counter_array.object_id
+    end
+
+    def perform(keys)
+      keys.each { |counter_array| fulfill(counter_array, counter_array[0]) }
+    end
+  end
+
+  class NilLoader < GraphQL::Dataloader::Source
+    def self.load
+      self.for().load(nil)
+    end
+
+    def perform(nils)
+      nils.each { |key| fulfill(nil, nil) }
+    end
+  end
+
+  class ImageType < GraphQL::Schema::Object
+    field :id, ID, null: false
+    field :filename, String, null: false
+  end
+
+  class ProductVariantType < GraphQL::Schema::Object
+    field :id, ID, null: false
+    field :title, String, null: false
+    field :image_ids, [ID, null: true], null: false
+
+    def image_ids
+      AssociationLoader.for(ProductVariant, :images).load(object).then do |images|
+        images.map(&:id)
+      end
+    end
+
+    field :product, GraphQL::Schema::LateBoundType.new('Product'), null: false
+
+    def product
+      RecordLoader.for(Product).load(object.product_id)
+    end
+  end
+
+  class ProductType < GraphQL::Schema::Object
+    field :id, ID, null: false
+    field :title, String, null: false
+    field :images, [ImageType], null: true
+
+    def images
+      product_image_query = RecordLoader.for(Image).load(object.image_id)
+      variant_images_query = AssociationLoader.for(Product, :variants).load(object).then do |variants|
+        variant_image_queries = variants.map do |variant|
+          AssociationLoader.for(ProductVariant, :images).load(variant)
+        end
+        GraphQL::Execution::Lazy.all(variant_image_queries).then(&:flatten)
+      end
+      GraphQL::Execution::Lazy.all([product_image_query, variant_images_query]).then do |product_image, variant_images|
+        # TODO this previously used `.value` to get inner values
+        [product_image] + variant_images
+      end
+    end
+
+    field :non_null_but_raises, String, null: false
+
+    def non_null_but_raises
+      raise GraphQL::ExecutionError, 'Error'
+    end
+
+    field :variants, [ProductVariantType], null: true
+
+    def variants
+      AssociationLoader.for(Product, :variants).load(object)
+    end
+
+    field :variants_count, Int, null: true
+
+    def variants_count
+      query = AssociationLoader.for(Product, :variants).load(object)
+      GraphQL::Execution::Lazy.all([query]).then { query.sync.size }
+    end
+  end
+
+  class QueryType < GraphQL::Schema::Object
+    field :constant, String, null: false
+
+    def constant
+      "constant value"
+    end
+
+    field :load_execution_error, String, null: true
+
+    def load_execution_error
+      RecordLoader.for(Product).load(1).then do |product|
+        raise GraphQL::ExecutionError, "test error message"
+      end
+    end
+
+    field :non_null_but_raises, ProductType, null: false
+
+    def non_null_but_raises
+      raise GraphQL::ExecutionError, 'Error'
+    end
+
+    field :non_null_but_promise_raises, String, null: false
+
+    def non_null_but_promise_raises
+      NilLoader.load.then do
+        raise GraphQL::ExecutionError, 'Error'
+      end
+    end
+
+    field :product, ProductType, null: true do
+      argument :id, ID, required: true
+    end
+
+    def product(id:)
+      RecordLoader.for(Product).load(id)
+    end
+
+    field :products, [ProductType], null: true do
+      argument :first, Int, required: true
+    end
+
+    def products(first:)
+      Product.first(first)
+    end
+
+    field :product_variants_count, Int, null: true do
+      argument :id, ID, required: true
+    end
+
+    def product_variants_count(id:)
+      RecordLoader.for(Product).load(id).then do |product|
+        AssociationLoader.for(Product, :variants).load(product).then(&:size)
+      end
+    end
+  end
+
+  class CounterType < GraphQL::Schema::Object
+    field :value, Int, null: false
+
+    def value
+      object
+    end
+
+    field :load_value, Int, null: false
+
+    def load_value
+      CounterLoader.load(context[:counter])
+    end
+  end
+
+  class IncrementCounterMutation < GraphQL::Schema::Mutation
+    null false
+    payload_type CounterType
+
+    def resolve
+      context[:counter][0] += 1
+      CounterLoader.load(context[:counter])
+    end
+  end
+
+  class CounterLoaderMutation < GraphQL::Schema::Mutation
+    null false
+    payload_type Int
+
+    def resolve
+      CounterLoader.load(context[:counter])
+    end
+  end
+
+  class NoOpMutation < GraphQL::Schema::Mutation
+    null false
+    payload_type QueryType
+
+    def resolve
+      Hash.new
+    end
+  end
+
+  class MutationType < GraphQL::Schema::Object
+    field :increment_counter, mutation: IncrementCounterMutation
+    field :counter_loader, mutation: CounterLoaderMutation
+    field :no_op, mutation: NoOpMutation
+  end
+
+  class Schema < GraphQL::Schema
+    query QueryType
+    mutation MutationType
+    use GraphQL::Dataloader
+  end
+
+  attr_reader :queries
+
+  def setup
+    @queries = []
+    QueryNotifier.subscriber = ->(query) { @queries << query }
+  end
+
+  def teardown
+    QueryNotifier.subscriber = nil
+  end
+
+  def test_no_queries
+    query_string = '{ constant }'
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "constant" => "constant value"
+      }
+    }
+    assert_equal expected, result
+    assert_equal [], queries
+  end
+
+  def test_single_query
+    query_string = <<-GRAPHQL
+      {
+        product(id: "1") {
+          id
+          title
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "product" => {
+          "id" => "1",
+          "title" => "Shirt",
+        }
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product/1"], queries
+  end
+
+  def test_batched_find_by_id
+    query_string = <<-GRAPHQL
+      {
+        product1: product(id: "1") { id, title }
+        product2: product(id: "2") { id, title }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "product1" => { "id" => "1", "title" => "Shirt" },
+        "product2" => { "id" => "2", "title" => "Pants" },
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product/1,2"], queries
+  end
+
+  def test_record_missing
+    query_string = <<-GRAPHQL
+      {
+        product(id: "123") {
+          id
+          title
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = { "data" => { "product" => nil } }
+    assert_equal expected, result
+    assert_equal ["Product/123"], queries
+  end
+
+  def test_non_null_field_that_raises_on_nullable_parent
+    query_string = <<-GRAPHQL
+      {
+        product(id: "1") {
+          id
+          nonNullButRaises
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = { 'data' => { 'product' => nil }, 'errors' => [{ 'message' => 'Error', 'locations' => [{ 'line' => 4, 'column' => 11 }], 'path' => ['product', 'nonNullButRaises'] }] }
+    assert_equal expected, result
+  end
+
+  def test_non_null_field_that_raises_on_query_root
+    query_string = <<-GRAPHQL
+      {
+        nonNullButRaises {
+          id
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = { 'data' => nil, 'errors' => [{ 'message' => 'Error', 'locations' => [{ 'line' => 2, 'column' => 9 }], 'path' => ['nonNullButRaises'] }] }
+    assert_equal expected, result
+  end
+
+  def test_non_null_field_promise_raises
+    result = Schema.execute('{ nonNullButPromiseRaises }')
+    expected = { 'data' => nil, 'errors' => [{ 'message' => 'Error', 'locations' => [{ 'line' => 1, 'column' => 3 }], 'path' => ['nonNullButPromiseRaises'] }] }
+    assert_equal expected, result
+  end
+
+  def test_batched_association_preload
+    query_string = <<-GRAPHQL
+      {
+        products(first: 2) {
+          id
+          title
+          variants {
+            id
+            title
+          }
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "products" => [
+          {
+            "id" => "1",
+            "title" => "Shirt",
+            "variants" => [
+              { "id" => "1", "title" => "Red" },
+              { "id" => "2", "title" => "Blue" },
+            ],
+          },
+          {
+            "id" => "2",
+            "title" => "Pants",
+            "variants" => [
+              { "id" => "4", "title" => "Small" },
+              { "id" => "5", "title" => "Medium" },
+              { "id" => "6", "title" => "Large" },
+            ],
+          }
+        ]
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product?limit=2", "Product/1,2/variants"], queries
+  end
+
+  def test_query_group_with_single_query
+    query_string = <<-GRAPHQL
+      {
+        products(first: 2) {
+          id
+          title
+          variantsCount
+          variants {
+            id
+            title
+          }
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "products" => [
+          {
+            "id" => "1",
+            "title" => "Shirt",
+            "variantsCount" => 2,
+            "variants" => [
+              { "id" => "1", "title" => "Red" },
+              { "id" => "2", "title" => "Blue" },
+            ],
+          },
+          {
+            "id" => "2",
+            "title" => "Pants",
+            "variantsCount" => 3,
+            "variants" => [
+              { "id" => "4", "title" => "Small" },
+              { "id" => "5", "title" => "Medium" },
+              { "id" => "6", "title" => "Large" },
+            ],
+          }
+        ]
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product?limit=2", "Product/1,2/variants"], queries
+  end
+
+  def test_sub_queries
+    query_string = <<-GRAPHQL
+      {
+        productVariantsCount(id: "2")
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "productVariantsCount" => 3
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product/2", "Product/2/variants"], queries
+  end
+
+  def test_query_group_with_sub_queries
+    query_string = <<-GRAPHQL
+      {
+        product(id: "1") {
+          images { id, filename }
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "product" => {
+          "images" => [
+            { "id" => "1", "filename" => "shirt.jpg" },
+            { "id" => "4", "filename" => "red-shirt.jpg" },
+            { "id" => "5", "filename" => "blue-shirt.jpg" },
+          ]
+        }
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product/1", "Image/1", "Product/1/variants", "ProductVariant/1,2/images"], queries
+  end
+
+  def test_load_list_of_objects_with_loaded_field
+    query_string = <<-GRAPHQL
+      {
+        products(first: 2) {
+          id
+          variants {
+            id
+            imageIds
+          }
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "products" => [
+          {
+            "id" => "1",
+            "variants" => [
+              { "id" => "1", "imageIds" => ["4"] },
+              { "id" => "2", "imageIds" => ["5"] },
+            ],
+          },
+          {
+            "id" => "2",
+            "variants" => [
+              { "id" => "4", "imageIds" => [] },
+              { "id" => "5", "imageIds" => [] },
+              { "id" => "6", "imageIds" => [] },
+            ],
+          }
+        ]
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product?limit=2", "Product/1,2/variants", "ProductVariant/1,2,4,5,6/images"], queries
+  end
+
+  def test_loader_reused_after_loading
+    query_string = <<-GRAPHQL
+      {
+        product(id: "2") {
+          variants {
+            id
+            product {
+              id
+              title
+            }
+          }
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "product" => {
+          "variants" => [
+            { "id" => "4", "product" => { "id" => "2", "title" => "Pants" } },
+            { "id" => "5", "product" => { "id" => "2", "title" => "Pants" } },
+            { "id" => "6", "product" => { "id" => "2", "title" => "Pants" } },
+          ],
+        }
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product/2", "Product/2/variants"], queries
+  end
+
+  def test_load_error
+    query_string = <<-GRAPHQL
+      {
+        constant
+        loadExecutionError
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => { "constant"=>"constant value", "loadExecutionError" => nil },
+      "errors" => [{ "message" => "test error message", "locations"=>[{"line"=>3, "column"=>9}], "path" => ["loadExecutionError"] }],
+    }
+    assert_equal expected, result
+  end
+
+  def test_mutation_execution
+    query_string = <<-GRAPHQL
+      mutation {
+        count1: counterLoader
+        incr1: incrementCounter { value, loadValue }
+        count2: counterLoader
+        incr2: incrementCounter { value, loadValue }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string, context: { counter: [0] })
+    expected = {
+      "data" => {
+        "count1" => 0,
+        "incr1" => { "value" => 1, "loadValue" => 1 },
+        "count2" => 1,
+        "incr2" => { "value" => 2, "loadValue" => 2 },
+      }
+    }
+    assert_equal expected, result
+  end
+
+  def test_mutation_batch_subselection_execution
+    query_string = <<-GRAPHQL
+      mutation {
+        mutation1: noOp {
+          product1: product(id: "1") { id, title }
+          product2: product(id: "2") { id, title }
+        }
+        mutation2: noOp {
+          product1: product(id: "2") { id, title }
+          product2: product(id: "3") { id, title }
+        }
+      }
+    GRAPHQL
+    result = Schema.execute(query_string)
+    expected = {
+      "data" => {
+        "mutation1" => {
+          "product1" => { "id" => "1", "title" => "Shirt" },
+          "product2" => { "id" => "2", "title" => "Pants" },
+        },
+        "mutation2" => {
+          "product1" => { "id" => "2", "title" => "Pants" },
+          "product2" => { "id" => "3", "title" => "Sweater" },
+        }
+      }
+    }
+    assert_equal expected, result
+    assert_equal ["Product/1,2", "Product/2,3"], queries
+  end
+end
diff --git a/spec/graphql/dataloader/redis_spec.rb b/spec/graphql/dataloader/redis_spec.rb
new file mode 100644
index 0000000000..5628251c31
--- /dev/null
+++ b/spec/graphql/dataloader/redis_spec.rb
@@ -0,0 +1,118 @@
+# frozen_string_literal: true
+require "spec_helper"
+
+describe GraphQL::Dataloader::Redis do
+  class DataloaderRedisSchema < GraphQL::Schema
+    class MockRedis
+      DATA = {
+        "l1" => [1,2,3],
+        "l2" => [4,5],
+        "k1" => "hello",
+        "k2" => "goodbye",
+      }
+
+      attr_reader :log
+      def initialize
+        @log = []
+      end
+
+      def pipelined
+        @new_log = []
+        res = yield
+        @log << @new_log
+        @new_log = nil
+        res
+      end
+
+      def scard(name)
+        @new_log << [:scard, name]
+        if (d = DATA[name])
+          if d.is_a?(Array)
+            d.size
+          else
+            raise "Invalid data type for #{name.inspect}"
+          end
+        else
+          0
+        end
+      end
+
+      def get(name)
+        @new_log << [:get, name]
+        if (d = DATA[name])
+          if d.is_a?(String)
+            d
+          else
+            raise "Invalid data type for #{name.inspect}"
+          end
+        else
+          nil
+        end
+      end
+    end
+
+    MOCK_REDIS = MockRedis.new
+
+    class Query < GraphQL::Schema::Object
+      field :count_list, Integer, null: false do
+        argument :name, String, required: true
+      end
+
+      def count_list(name:)
+        GraphQL::Dataloader::Redis.load(MOCK_REDIS, [:scard, name])
+      end
+
+      field :get_string, String, null: true do
+        argument :name, String, required: true
+      end
+
+      def get_string(name:)
+        GraphQL::Dataloader::Redis.load(MOCK_REDIS, [:get, name])
+      end
+    end
+
+    query(Query)
+    use GraphQL::Dataloader
+  end
+
+  def exec_query(*args, **kwargs)
+    DataloaderRedisSchema.execute(*args, **kwargs)
+  end
+
+  before do
+    DataloaderRedisSchema::MOCK_REDIS.log.clear
+  end
+
+  it "dispatches to .pipeline and sends method calls" do
+    res = exec_query <<-GRAPHQL
+    {
+      k1: getString(name: "k1")
+      l1: countList(name: "l1")
+      k2: getString(name: "k2")
+      l5: countList(name: "l5")
+      k3: getString(name: "k3")
+    }
+    GRAPHQL
+
+    expected_log = [
+      [
+        [:get, "k1"],
+        [:scard, "l1"],
+        [:get, "k2"],
+        [:scard, "l5"],
+        [:get, "k3"]
+      ]
+    ]
+
+    assert_equal expected_log, DataloaderRedisSchema::MOCK_REDIS.log
+
+    expected_data = {
+      "k1" => "hello",
+      "l1" => 3,
+      "k2" => "goodbye",
+      "l5" => 0,
+      "k3" => nil
+    }
+    assert_equal(expected_data, res["data"])
+  end
+end
diff --git a/spec/graphql/dataloader_spec.rb b/spec/graphql/dataloader_spec.rb
new file mode 100644
index 0000000000..da1b7af886
--- /dev/null
+++ b/spec/graphql/dataloader_spec.rb
@@ -0,0 +1,330 @@
+# frozen_string_literal: true
+require "spec_helper"
+
+describe "GraphQL::Dataloader" do
+  module DataloaderTest
+    module Backend
+      LOG = []
+      DEFAULT_DATA = {
+        "b1" => { title: "Remembering", author_id: "a1" },
+        "b2" => { title: "That Distant Land", author_id: "a1" },
+        "b3" => { title: "Doggies", author_id: "a2" },
+        "b4" => { title: "The Cloud of Unknowing", author_id: "a3" }, # This author intentionally missing
+        "a1" => { name: "Wendell Berry", book_ids: ["b1", "b2"] },
+        "a2" => { name: "Sandra Boynton", book_ids: ["b3"] },
+      }.freeze
+
+      def self.reset
+        self.data = DEFAULT_DATA.dup
+      end
+
+      class << self
+        attr_accessor :data
+      end
+
+      def self.mget(keys)
+        LOG << "MGET #{keys}"
+        keys.map { |k| self.data[k] || raise("Key not found: #{k}") }
+      end
+
+      def self.set(id, object)
+        self.data[id] = object
+      end
+    end
+
+    class BackendSource < GraphQL::Dataloader::Source
+      def initialize(key)
+        @key = key
+      end
+
+      def self.load_object(id)
+        load(nil, id)
+      end
+
+      def perform(ids)
+        ids = ids.sort # for stable logging
+        Backend.mget(ids).each_with_index do |item, idx|
+          fulfill(ids[idx], item)
+        end
+      end
+    end
+
+    class BackgroundThreadBackendSource < BackendSource
+      include GraphQL::Dataloader::Source::BackgroundThreaded
+
+      def perform(ids)
+        sleep 0.5
+        super
+      end
+    end
+
+    class Schema < GraphQL::Schema
+      class BaseObject < GraphQL::Schema::Object
+        private
+
+        def source_class
+          @context[:background_threaded] ? BackgroundThreadBackendSource : BackendSource
+        end
+      end
+
+      class Author < BaseObject
+        field :name, String, null: false
+        field :books, [GraphQL::Schema::LateBoundType.new("Book")], null: false
+
+        def books
+          source_class.load_all(nil, object[:book_ids])
+        end
+      end
+
+      class Book < BaseObject
+        field :title, String, null: false
+        field :author, Author, null: false
+
+        def author
+          source_class.load_object(object[:author_id])
+        end
+      end
+
+      class ObjectUnion < GraphQL::Schema::Union
+        possible_types Book, Author
+      end
+
+      class Query < BaseObject
+        field :book, Book, null: true do
+          argument :id, ID, required: true
+        end
+
+        def book(id:)
+          source_class.load_object(id)
+        end
+
+        field :author, Author, null: true do
+          argument :id, ID, required: true
+        end
+
+        def author(id:)
+          source_class.load_object(id)
+        end
+
+        field :books_count, Integer, null: false do
+          argument :author_id, ID, required: true
+        end
+
+        def books_count(author_id:)
+          # Of course this could be done without a nested load, but I want to test nested loaders
+          source_class.load_object(author_id).then do |author|
+            source_class.load_all(nil, author[:book_ids]).then do |books|
+              books.size
+            end
+          end
+        end
+
+        field :object, ObjectUnion, null: true, resolver_method: :load_object do
+          argument :type, String, required: true
+          argument :id, ID, required: true
+        end
+
+        def load_object(type:, id:)
+          source_class.load(type, id)
+        end
+      end
+
+      class Mutation < BaseObject
+        field :add_author, Author, null: true do
+          argument :id, ID, required: true
+          argument :name, String, required: true
+          argument :book_ids, [ID], required: true
+        end
+
+        def add_author(id:, name:, book_ids:)
+          author = { name: name, book_ids: book_ids }
+          Backend.set(id, author)
+          author
+        end
+      end
+
+      query(Query)
+      mutation(Mutation)
+      use GraphQL::Dataloader
+
+      def self.resolve_type(type, obj, ctx)
+        if obj.key?(:name)
+          Author
+        elsif obj.key?(:title)
+          Book
+        else
+          raise "Unknown object: #{obj.inspect}"
+        end
+      end
+    end
+  end
+
+  def exec_query(*args, **kwargs)
+    DataloaderTest::Schema.execute(*args, **kwargs)
+  end
+
+  let(:log) { DataloaderTest::Backend::LOG }
+
+  before do
+    DataloaderTest::Backend.reset
+    log.clear
+  end
+
+  it "batches requests" do
+    res = exec_query('{
+      b1: book(id: "b1") { title author { name } }
+      b2: book(id: "b2") { title author { name } }
+    }')
+
+    assert_equal "Remembering", res["data"]["b1"]["title"]
+    assert_equal "Wendell Berry", res["data"]["b1"]["author"]["name"]
+    assert_equal "That Distant Land", res["data"]["b2"]["title"]
+    assert_equal "Wendell Berry", res["data"]["b2"]["author"]["name"]
+    assert_equal ['MGET ["b1", "b2"]', 'MGET ["a1"]'], log
+  end
+
+  it "batches requests across branches of a query" do
+    exec_query('{
+      a1: author(id: "a1") { books { title } }
+      a2: author(id: "a2") { books { title } }
+    }')
+
+    expected_log = [
+      "MGET [\"a1\", \"a2\"]",
+      "MGET [\"b1\", \"b2\", \"b3\"]"
+    ]
+    assert_equal expected_log, log
+  end
+
+  it "doesn't load the same object over again" do
+    exec_query('{
+      b1: book(id: "b1") {
+        title
+        author { name }
+      }
+      a1: author(id: "a1") {
+        books {
+          author {
+            books {
+              title
+            }
+          }
+        }
+      }
+    }')
+
+    expected_log = [
+      'MGET ["a1", "b1"]',
+      'MGET ["b2"]'
+    ]
+    assert_equal expected_log, log
+  end
+
+  it "shares over a multiplex" do
+    query_string = "query($id: ID!) { author(id: $id) { name } }"
+    results = DataloaderTest::Schema.multiplex([
+      { query: query_string, variables: { "id" => "a1" } },
+      { query: query_string, variables: { "id" => "a2" } },
+    ])
+
+    assert_equal "Wendell Berry", results[0]["data"]["author"]["name"]
+    assert_equal "Sandra Boynton", results[1]["data"]["author"]["name"]
+    assert_equal ["MGET [\"a1\", \"a2\"]"], log
+  end
+
+  it "doesn't batch between mutations" do
+    query_str = <<-GRAPHQL
+      mutation {
+        add1: addAuthor(id: "a3", name: "Beatrix Potter", bookIds: ["b1", "b2"]) {
+          books {
+            title
+          }
+        }
+        add2: addAuthor(id: "a4", name: "Joel Salatin", bookIds: ["b1", "b3"]) {
+          books {
+            title
+          }
+        }
+      }
+    GRAPHQL
+
+    exec_query(query_str)
+    expected_log = ['MGET ["b1", "b2"]', 'MGET ["b1", "b3"]']
+    assert_equal expected_log, log
+  end
+
+  it "works with nested sources" do
+    query_str = <<-GRAPHQL
+    {
+      a1: booksCount(authorId: "a1")
+      a2: booksCount(authorId: "a2")
+    }
+    GRAPHQL
+
+    res = exec_query(query_str)
+    assert_equal({"data"=>{"a1"=>2, "a2"=>1}}, res)
+    expected_log = [
+      'MGET ["a1", "a2"]',
+      'MGET ["b1", "b2", "b3"]',
+    ]
+    assert_equal expected_log, log
+  end
+
+  it "raises helpful errors" do
+    err = assert_raises GraphQL::Dataloader::LoadError do
+      exec_query('query GetBook { book4: book(id: "b4") { author { name } } }')
+    end
+    assert_equal "Key not found: a3", err.cause.message
+    assert_equal "Error from DataloaderTest::BackendSource#perform(\"a3\") at GetBook.book4.author\n\nRuntimeError:\n\"Key not found: a3\"", err.message
+    assert_equal ["book4", "author"], err.graphql_path
+  end
+
+  it "runs background thread loads in parallel" do
+    query_str = <<-GRAPHQL
+    {
+      o1: object(type: "Author", id: "a1") { ... AuthorFields }
+      o2: object(type: "Author", id: "a2") { ... AuthorFields }
+      o3: object(type: "Book", id: "b1") { ... BookFields }
+    }
+
+    fragment AuthorFields on Author {
+      name
+    }
+    fragment BookFields on Book {
+      title
+    }
+    GRAPHQL
+    started_at = Time.now
+    res = exec_query(query_str, context: { background_threaded: true })
+    ended_at = Time.now
+    expected_data = {"o1"=>{"name"=>"Wendell Berry"}, "o2"=>{"name"=>"Sandra Boynton"}, "o3"=>{"title"=>"Remembering"}}
+    assert_equal(expected_data, res["data"])
+    assert_in_delta 0.5, ended_at - started_at, 0.1
+  end
+
+  it "raises helpful errors from background threads" do
+    err = assert_raises GraphQL::Dataloader::LoadError do
+      exec_query('query GetBook { book4: book(id: "b4") { author { name } } }', context: { background_threaded: true })
+    end
+    assert_equal "Key not found: a3", err.cause.message
+    assert_equal "Error from DataloaderTest::BackgroundThreadBackendSource#perform(\"a3\") at GetBook.book4.author\n\nRuntimeError:\n\"Key not found: a3\"", err.message
+    assert_equal ["book4", "author"], err.graphql_path
+  end
+
+  it "works with backgrounded, nested sources" do
+    query_str = <<-GRAPHQL
+    {
+      a1: booksCount(authorId: "a1")
+      a2: booksCount(authorId: "a2")
+    }
+    GRAPHQL
+
+    res = exec_query(query_str, context: { background_threaded: true })
+    assert_equal({"data"=>{"a1"=>2, "a2"=>1}}, res)
+    expected_log = [
+      'MGET ["a1", "a2"]',
+      'MGET ["b1", "b2", "b3"]',
+    ]
+    assert_equal expected_log, log
+  end
+end
diff --git a/spec/support/active_record_setup.rb b/spec/support/active_record_setup.rb
index 523a9430f3..57c54d8a7c 100644
--- a/spec/support/active_record_setup.rb
+++ b/spec/support/active_record_setup.rb
@@ -46,5 +46,19 @@ def jruby?
     create_table :foods, force: true do |t|
       t.column :name, :string
     end
+
+    create_table :albums, force: true do |t|
+      t.column :name, :string
+      t.column :artist_id, :integer
+    end
+
+    create_table :artists, force: true do |t|
+      t.column :name, :string
+    end
+
+    create_table :html_colors, force: true do |t|
+      t.column :name, :string
+      t.column :hex, :integer
+    end
   end
 end