diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6e6b67c..803af73 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,7 @@ Please mark backwards incompatible changes with an exclamation mark at the start
 ## [Unreleased]
 
 ### Added
+- The `_source` parameter for Elasticsearch's `TopHits` aggregation.
 - The `sort` parameter for Elasticsearch's `TopHits` aggregation.
 - The `missing` parameter for Elasticsearch's `Terms` aggregation.
 
diff --git a/lib/jay_api/elasticsearch/query_builder/aggregations/top_hits.rb b/lib/jay_api/elasticsearch/query_builder/aggregations/top_hits.rb
index 4715e86..2f10596 100644
--- a/lib/jay_api/elasticsearch/query_builder/aggregations/top_hits.rb
+++ b/lib/jay_api/elasticsearch/query_builder/aggregations/top_hits.rb
@@ -15,7 +15,7 @@ class Aggregations
         # Information on this type of aggregation can be found here:
         # https://www.elastic.co/docs/reference/aggregations/search-aggregations-metrics-top-hits-aggregation
         class TopHits < ::JayAPI::Elasticsearch::QueryBuilder::Aggregations::Aggregation
-          attr_reader :size, :sort
+          attr_reader :size, :sort, :_source
 
           # @param [String] name The name used by Elasticsearch to identify each
           #   of the aggregations.
@@ -23,16 +23,23 @@ class TopHits < ::JayAPI::Elasticsearch::QueryBuilder::Aggregations::Aggregation
           # @param [Hash] sort How to sort the documents to determine the top
           #   hits. When not specified the documents are sorted by the score of
           #   the main query.
-          def initialize(name, size:, sort: nil)
+          # @param [FalseClass, String, Array<String>, Hash] _source Expression
+          #   used for source filtering.
+          # @see https://www.elastic.co/guide/en/elasticsearch/reference/current/search-fields.html#source-filtering
+          #   Elasticsearch's documentation for more information on what kind of
+          #   expressions are allowed.
+          # rubocop:disable-next Lint/UnderscorePrefixedVariableName -- to match Elasticsearch's name
+          def initialize(name, size:, sort: nil, _source: nil)
             super(name)
 
             @size = size
             @sort = sort
+            @_source = _source
           end
 
           # @return [self] A copy of the receiver.
           def clone
-            optional_params = { sort: sort.deep_dup }.compact
+            optional_params = { sort: sort.deep_dup, _source: _source.deep_dup }.compact
             copy = self.class.new(name, size: size, **optional_params)
             copy.aggregations = aggregations.clone
             copy
@@ -45,7 +52,8 @@ def to_h
               {
                 top_hits: {
                   size: size,
-                  sort: sort.deep_dup
+                  sort: sort.deep_dup,
+                  _source: _source.deep_dup
                 }.compact
               }
             end
diff --git a/spec/jay_api/elasticsearch/query_builder/aggregations/top_hits_spec.rb b/spec/jay_api/elasticsearch/query_builder/aggregations/top_hits_spec.rb
index 877d257..0a4a160 100644
--- a/spec/jay_api/elasticsearch/query_builder/aggregations/top_hits_spec.rb
+++ b/spec/jay_api/elasticsearch/query_builder/aggregations/top_hits_spec.rb
@@ -47,6 +47,50 @@
       end
     end
 
+    shared_examples_for "#clone when a '_source' has been given" do
+      it "has its '_source' set to the same value" do
+        expect(method_call._source).to eq(_source)
+      end
+    end
+
+    shared_examples_for "#clone when a non-scalar '_source' has been given" do
+      it_behaves_like "#clone when a '_source' has been given"
+
+      it 'is not the same object' do
+        expect(method_call._source).not_to be(top_hits._source)
+      end
+    end
+
+    context "when a '_source' has been given" do
+      let(:constructor_params) { super().merge(_source:) }
+
+      context "when '_source' is a boolean" do
+        let(:_source) { false }
+
+        it_behaves_like "#clone when a '_source' has been given"
+      end
+
+      context "when '_source' is a string" do
+        let(:_source) { 'meta_data.*' }
+
+        it_behaves_like "#clone when a non-scalar '_source' has been given"
+      end
+
+      context "when '_source' is an array" do
+        let(:_source) { %w[meta_data.* timestamp] }
+
+        it_behaves_like "#clone when a non-scalar '_source' has been given"
+      end
+
+      context "when '_source' is a hash" do
+        let(:_source) do
+          { includes: %i[date price] }
+        end
+
+        it_behaves_like "#clone when a non-scalar '_source' has been given"
+      end
+    end
+
     context 'when the original object has nested aggregations' do
       let(:cloned_aggregations) { instance_double(JayAPI::Elasticsearch::QueryBuilder::Aggregations) }
 
@@ -110,5 +154,53 @@
         expect(sort_hash).to eq(sort)
       end
     end
+
+    context "when a '_source' has been given" do
+      let(:constructor_params) { super().merge(_source:) }
+
+      context "when '_source' is a scalar" do
+        let(:_source) { false }
+
+        let(:expected_hash) do
+          {
+            'an_aggregation_sample' => {
+              top_hits: {
+                size: 1,
+                _source: false
+              }
+            }
+          }
+        end
+
+        it 'returns the expected Hash' do
+          expect(method_call).to eq(expected_hash)
+        end
+      end
+
+      context "when '_source' is not a scalar" do
+        let(:_source) { %w[meta_data.* timestamp] }
+
+        let(:expected_hash) do
+          {
+            'an_aggregation_sample' => {
+              top_hits: {
+                size: 1,
+                _source: %w[meta_data.* timestamp]
+              }
+            }
+          }
+        end
+
+        it 'returns the expected Hash' do
+          expect(method_call).to eq(expected_hash)
+        end
+
+        it "does not return a reference to the internal '_source' object" do
+          source_hash = method_call.dig('an_aggregation_sample', :top_hits, :_source)
+          expect(source_hash).not_to be(top_hits._source)
+          expect(source_hash).to eq(_source)
+        end
+      end
+    end
   end
 end