diff --git a/esrally/track/loader.py b/esrally/track/loader.py
index 39800a6a..fbe3a076 100644
--- a/esrally/track/loader.py
+++ b/esrally/track/loader.py
@@ -228,6 +228,19 @@ def operation_parameters(t, op):
         return params.param_source_for_operation(op.type, t, op.params)
 
 
+def used_corpora(t, cfg):
+    corpora = {}
+    challenge = t.find_challenge_or_default(cfg.opts("track", "challenge.name"))
+    for task in challenge.schedule:
+        for sub_task in task:
+            param_source = operation_parameters(t, sub_task.operation)
+            if hasattr(param_source, "corpora"):
+                for c in param_source.corpora:
+                    # We might have the same corpus *but* they contain different doc sets. Therefore also need to union over doc sets.
+                    corpora[c.name] = corpora.get(c.name, c).union(c)
+    return corpora.values()
+
+
 def prepare_track(t, cfg):
     """
     Ensures that all track data are available for running the benchmark.
@@ -238,7 +251,7 @@ def prepare_track(t, cfg):
     logger = logging.getLogger(__name__)
     offline = cfg.opts("system", "offline.mode")
     test_mode = cfg.opts("track", "test.mode.enabled")
-    for corpus in t.corpora:
+    for corpus in used_corpora(t, cfg):
         data_root = data_dir(cfg, t.name, corpus.name)
         logger.info("Resolved data root directory for document corpus [%s] in track [%s] to %s.", corpus.name, t.name, data_root)
         prep = DocumentSetPreparator(t.name, offline, test_mode)
diff --git a/esrally/track/track.py b/esrally/track/track.py
index dcdbe72b..68083405 100644
--- a/esrally/track/track.py
+++ b/esrally/track/track.py
@@ -214,6 +214,14 @@ class DocumentCorpus:
             filtered.append(d)
         return DocumentCorpus(self.name, filtered)
 
+    def union(self, other):
+        if self.name != other.name:
+            raise exceptions.RallyAssertionError("Both document corpora must have the same name")
+        if self is other:
+            return self
+        else:
+            return DocumentCorpus(self.name, list(set(self.documents).union(other.documents)))
+
     def __str__(self):
         return self.name
 

Only load required corpora
Currently Rally is loading all referenced corpora of track, even if they are not needed. As it is possible that we download and extract a significant amount of data (usually several GB if not more), we should investigate whether we can load only the required ones.
diff --git a/tests/track/loader_test.py b/tests/track/loader_test.py
index e487640a..09e29262 100644
--- a/tests/track/loader_test.py
+++ b/tests/track/loader_test.py
@@ -459,6 +459,140 @@ class TrackPreparationTests(TestCase):
         self.assertEqual(0, decompress.call_count)
         self.assertEqual(0, prepare_file_offset_table.call_count)
 
+    def test_used_corpora(self):
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "track", "challenge.name", "default-challenge")
+        track_specification = {
+            "description": "description for unit test",
+            "indices": [
+                {"name": "logs-181998"},
+                {"name": "logs-191998"},
+                {"name": "logs-201998"},
+            ],
+            "corpora": [
+                {
+                    "name": "http_logs_unparsed",
+                    "target-type": "type",
+                    "documents": [
+                        {
+                            "target-index": "logs-181998",
+                            "source-file": "documents-181998.unparsed.json.bz2",
+                            "document-count": 2708746,
+                            "compressed-bytes": 13064317,
+                            "uncompressed-bytes": 303920342
+                        },
+                        {
+                            "target-index": "logs-191998",
+                            "source-file": "documents-191998.unparsed.json.bz2",
+                            "document-count": 9697882,
+                            "compressed-bytes": 47211781,
+                            "uncompressed-bytes": 1088378738
+                        },
+                        {
+                            "target-index": "logs-201998",
+                            "source-file": "documents-201998.unparsed.json.bz2",
+                            "document-count": 13053463,
+                            "compressed-bytes": 63174979,
+                            "uncompressed-bytes": 1456836090
+                        }
+                    ]
+                },
+                {
+                    "name": "http_logs",
+                    "target-type": "type",
+                    "documents": [
+                        {
+                            "target-index": "logs-181998",
+                            "source-file": "documents-181998.json.bz2",
+                            "document-count": 2708746,
+                            "compressed-bytes": 13815456,
+                            "uncompressed-bytes": 363512754
+                        },
+                        {
+                            "target-index": "logs-191998",
+                            "source-file": "documents-191998.json.bz2",
+                            "document-count": 9697882,
+                            "compressed-bytes": 49439633,
+                            "uncompressed-bytes": 1301732149
+                        },
+                        {
+                            "target-index": "logs-201998",
+                            "source-file": "documents-201998.json.bz2",
+                            "document-count": 13053463,
+                            "compressed-bytes": 65623436,
+                            "uncompressed-bytes": 1744012279
+                        }
+                    ]
+                }
+            ],
+            "operations": [
+                {
+                    "name": "bulk-index-1",
+                    "operation-type": "bulk",
+                    "corpora": ["http_logs"],
+                    "indices": ["logs-181998"],
+                    "bulk-size": 500
+                },
+                {
+                    "name": "bulk-index-2",
+                    "operation-type": "bulk",
+                    "corpora": ["http_logs"],
+                    "indices": ["logs-191998"],
+                    "bulk-size": 500
+                },
+                {
+                    "name": "bulk-index-3",
+                    "operation-type": "bulk",
+                    "corpora": ["http_logs_unparsed"],
+                    "indices": ["logs-201998"],
+                    "bulk-size": 500
+                },
+                {
+                    "name": "node-stats",
+                    "operation-type": "node-stats"
+                },
+            ],
+            "challenges": [
+                {
+                    "name": "default-challenge",
+                    "schedule": [
+                        {
+                            "parallel": {
+                                "tasks": [
+                                    {
+                                        "name": "index-1",
+                                        "operation": "bulk-index-1",
+                                    },
+                                    {
+                                        "name": "index-2",
+                                        "operation": "bulk-index-2",
+                                    },
+                                    {
+                                        "name": "index-3",
+                                        "operation": "bulk-index-3",
+                                    },
+                                ]
+                            }
+                        },
+                        {
+                            "operation": "node-stats"
+                        }
+                    ]
+                }
+            ]
+        }
+        reader = loader.TrackSpecificationReader()
+        full_track = reader("unittest", track_specification, "/mappings")
+        used_corpora = sorted(loader.used_corpora(full_track, cfg), key=lambda c: c.name)
+        self.assertEqual(2, len(used_corpora))
+        self.assertEqual("http_logs", used_corpora[0].name)
+        # each bulk operation requires a different data file but they should have been merged properly.
+        self.assertEqual({"documents-181998.json.bz2", "documents-191998.json.bz2"},
+                         {d.document_archive for d in used_corpora[0].documents})
+
+        self.assertEqual("http_logs_unparsed", used_corpora[1].name)
+        self.assertEqual({"documents-201998.unparsed.json.bz2"}, {d.document_archive for d in used_corpora[1].documents})
+
     @mock.patch("esrally.utils.io.prepare_file_offset_table")
     @mock.patch("esrally.utils.io.decompress")
     @mock.patch("os.path.getsize")
diff --git a/tests/track/track_test.py b/tests/track/track_test.py
index 05fa3907..bddd5d40 100644
--- a/tests/track/track_test.py
+++ b/tests/track/track_test.py
@@ -126,3 +126,32 @@ class DocumentCorpusTests(TestCase):
         self.assertEqual(2, len(filtered_corpus.documents))
         self.assertEqual("logs-01", filtered_corpus.documents[0].target_index)
         self.assertEqual("logs-02", filtered_corpus.documents[1].target_index)
+
+    def test_union_document_corpus_is_reflexive(self):
+        corpus = track.DocumentCorpus("test", documents=[
+            track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK, number_of_documents=5, target_index="logs-01"),
+            track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK, number_of_documents=6, target_index="logs-02"),
+            track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK, number_of_documents=7, target_index="logs-03"),
+            track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK, number_of_documents=8, target_index=None)
+        ])
+        self.assertTrue(corpus.union(corpus) is corpus)
+
+    def test_union_document_corpora_is_symmetric(self):
+        a = track.DocumentCorpus("test", documents=[
+            track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK, number_of_documents=5, target_index="logs-01"),
+        ])
+        b = track.DocumentCorpus("test", documents=[
+            track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK, number_of_documents=5, target_index="logs-02"),
+        ])
+        self.assertEqual(b.union(a), a.union(b))
+        self.assertEqual(2, len(a.union(b).documents))
+
+    def test_cannot_union_mixed_document_corpora(self):
+        a = track.DocumentCorpus("test", documents=[
+            track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK, number_of_documents=5, target_index="logs-01"),
+        ])
+        b = track.DocumentCorpus("other", documents=[
+            track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK, number_of_documents=5, target_index="logs-02"),
+        ])
+        with self.assertRaisesRegex(exceptions.RallyAssertionError, "Both document corpora must have the same name"):
+            a.union(b)

pip install -e .
parse_log_pytest
pytest
pytest pytest-benchmark
apt-get update apt-get install -y gcc
3.6
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs @ file:///opt/conda/conda-bld/attrs_1642510447205/work
certifi==2021.5.30
elasticsearch==6.2.0
-e git+https://github.com/elastic/rally.git@31ac8b0f9eace24cafff7d995daa1285d1443309#egg=esrally
importlib-metadata @ file:///tmp/build/80754af9/importlib-metadata_1631916693255/work
iniconfig @ file:///home/linux1/recipes/ci/iniconfig_1610983019677/work
Jinja2==2.9.5
jsonschema==2.5.1
MarkupSafe==2.0.1
more-itertools @ file:///tmp/build/80754af9/more-itertools_1637733554872/work
packaging @ file:///tmp/build/80754af9/packaging_1637314298585/work
pluggy @ file:///tmp/build/80754af9/pluggy_1615976315926/work
psutil==5.4.0
py @ file:///opt/conda/conda-bld/py_1644396412707/work
py-cpuinfo==3.2.0
pyparsing @ file:///tmp/build/80754af9/pyparsing_1635766073266/work
pytest==6.2.4
pytest-benchmark==3.4.1
tabulate==0.8.1
thespian==3.9.2
toml @ file:///tmp/build/80754af9/toml_1616166611790/work
typing_extensions @ file:///opt/conda/conda-bld/typing_extensions_1647553014482/work
urllib3==1.22
zipp @ file:///tmp/build/80754af9/zipp_1633618647012/work

name: rally
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - attrs=21.4.0=pyhd3eb1b0_0
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - importlib-metadata=4.8.1=py36h06a4308_0
  - importlib_metadata=4.8.1=hd3eb1b0_0
  - iniconfig=1.1.1=pyhd3eb1b0_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - more-itertools=8.12.0=pyhd3eb1b0_0
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - packaging=21.3=pyhd3eb1b0_0
  - pip=21.2.2=py36h06a4308_0
  - pluggy=0.13.1=py36h06a4308_0
  - py=1.11.0=pyhd3eb1b0_0
  - pyparsing=3.0.4=pyhd3eb1b0_0
  - pytest=6.2.4=py36h06a4308_2
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - toml=0.10.2=pyhd3eb1b0_0
  - typing_extensions=4.1.1=pyh06a4308_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zipp=3.6.0=pyhd3eb1b0_0
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - elasticsearch==6.2.0
      - jinja2==2.9.5
      - jsonschema==2.5.1
      - markupsafe==2.0.1
      - psutil==5.4.0
      - py-cpuinfo==3.2.0
      - pytest-benchmark==3.4.1
      - tabulate==0.8.1
      - thespian==3.9.2
      - urllib3==1.22
prefix: /opt/conda/envs/rally

tests/track/loader_test.py::TrackPreparationTests::test_used_corpora tests/track/track_test.py::DocumentCorpusTests::test_cannot_union_mixed_document_corpora tests/track/track_test.py::DocumentCorpusTests::test_union_document_corpora_is_symmetric tests/track/track_test.py::DocumentCorpusTests::test_union_document_corpus_is_reflexive
tests/track/loader_test.py::SimpleTrackRepositoryTests::test_track_from_directory tests/track/loader_test.py::SimpleTrackRepositoryTests::test_track_from_directory_without_track tests/track/loader_test.py::SimpleTrackRepositoryTests::test_track_from_file tests/track/loader_test.py::SimpleTrackRepositoryTests::test_track_from_file_but_not_json tests/track/loader_test.py::SimpleTrackRepositoryTests::test_track_from_named_pipe tests/track/loader_test.py::SimpleTrackRepositoryTests::test_track_from_non_existing_path tests/track/loader_test.py::GitRepositoryTests::test_track_from_existing_repo tests/track/loader_test.py::TrackPreparationTests::test_decompresses_if_archive_available tests/track/loader_test.py::TrackPreparationTests::test_does_nothing_if_document_file_available tests/track/loader_test.py::TrackPreparationTests::test_download_document_archive_if_no_file_available tests/track/loader_test.py::TrackPreparationTests::test_download_document_file_if_no_file_available tests/track/loader_test.py::TrackPreparationTests::test_prepare_bundled_document_set_decompresses_compressed_docs tests/track/loader_test.py::TrackPreparationTests::test_prepare_bundled_document_set_does_nothing_if_no_document_files tests/track/loader_test.py::TrackPreparationTests::test_prepare_bundled_document_set_error_compressed_docs_wrong_size tests/track/loader_test.py::TrackPreparationTests::test_prepare_bundled_document_set_if_document_file_available tests/track/loader_test.py::TrackPreparationTests::test_prepare_bundled_document_set_uncompressed_docs_wrong_size tests/track/loader_test.py::TrackPreparationTests::test_raise_download_error_if_no_url_provided_and_file_missing tests/track/loader_test.py::TrackPreparationTests::test_raise_download_error_if_no_url_provided_and_wrong_file_size tests/track/loader_test.py::TrackPreparationTests::test_raise_download_error_if_offline tests/track/loader_test.py::TrackPreparationTests::test_raise_download_error_no_test_mode_file tests/track/loader_test.py::TrackPreparationTests::test_raise_download_error_on_connection_problems tests/track/loader_test.py::TrackPreparationTests::test_raise_error_if_compressed_does_not_contain_expected_document_file tests/track/loader_test.py::TrackPreparationTests::test_raise_error_on_wrong_uncompressed_file_size tests/track/loader_test.py::TemplateRenderTests::test_render_simple_template tests/track/loader_test.py::TemplateRenderTests::test_render_template_with_external_variables tests/track/loader_test.py::TemplateRenderTests::test_render_template_with_globbing tests/track/loader_test.py::TemplateRenderTests::test_render_template_with_variables tests/track/loader_test.py::TrackPostProcessingTests::test_post_processes_track_spec tests/track/loader_test.py::TrackPathTests::test_sets_absolute_path tests/track/loader_test.py::TrackFilterTests::test_create_filters_from_empty_included_tasks tests/track/loader_test.py::TrackFilterTests::test_create_filters_from_mixed_included_tasks tests/track/loader_test.py::TrackFilterTests::test_filters_tasks tests/track/loader_test.py::TrackFilterTests::test_rejects_invalid_syntax tests/track/loader_test.py::TrackFilterTests::test_rejects_unknown_filter_type tests/track/loader_test.py::TrackSpecificationReaderTests::test_at_least_one_default_challenge tests/track/loader_test.py::TrackSpecificationReaderTests::test_auto_generates_challenge_from_schedule tests/track/loader_test.py::TrackSpecificationReaderTests::test_can_read_track_info tests/track/loader_test.py::TrackSpecificationReaderTests::test_description_is_optional tests/track/loader_test.py::TrackSpecificationReaderTests::test_document_count_mandatory_if_file_present tests/track/loader_test.py::TrackSpecificationReaderTests::test_exactly_one_default_challenge tests/track/loader_test.py::TrackSpecificationReaderTests::test_inline_operations tests/track/loader_test.py::TrackSpecificationReaderTests::test_not_more_than_one_default_challenge_possible tests/track/loader_test.py::TrackSpecificationReaderTests::test_parallel_tasks_with_completed_by_set tests/track/loader_test.py::TrackSpecificationReaderTests::test_parallel_tasks_with_completed_by_set_multiple_tasks_match tests/track/loader_test.py::TrackSpecificationReaderTests::test_parallel_tasks_with_completed_by_set_no_task_matches tests/track/loader_test.py::TrackSpecificationReaderTests::test_parallel_tasks_with_default_clients_does_not_propagate tests/track/loader_test.py::TrackSpecificationReaderTests::test_parallel_tasks_with_default_values tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_challenge_and_challenges_are_defined tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_duplicate_explicit_task_names tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_duplicate_implicit_task_names tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_missing_challenge_or_challenges tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_unique_task_names tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_valid_track_specification tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_valid_track_specification_with_index_template tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_with_mixed_warmup_iterations_and_measurement tests/track/loader_test.py::TrackSpecificationReaderTests::test_parse_with_mixed_warmup_time_period_and_iterations tests/track/loader_test.py::TrackSpecificationReaderTests::test_selects_sole_challenge_implicitly_as_default tests/track/loader_test.py::TrackSpecificationReaderTests::test_supports_target_interval tests/track/loader_test.py::TrackSpecificationReaderTests::test_supports_target_throughput tests/track/loader_test.py::TrackSpecificationReaderTests::test_unique_challenge_names tests/track/track_test.py::TrackTests::test_default_challenge_none_if_no_challenges tests/track/track_test.py::TrackTests::test_does_not_find_unknown_challenge tests/track/track_test.py::TrackTests::test_finds_challenge_by_name tests/track/track_test.py::TrackTests::test_finds_default_challenge tests/track/track_test.py::TrackTests::test_uses_default_challenge_if_no_name_given tests/track/track_test.py::IndexTests::test_matches_exactly tests/track/track_test.py::IndexTests::test_matches_if_catch_all_pattern_is_defined tests/track/track_test.py::IndexTests::test_matches_if_no_pattern_is_defined tests/track/track_test.py::IndexTests::test_str tests/track/track_test.py::DocumentCorpusTests::test_do_not_filter tests/track/track_test.py::DocumentCorpusTests::test_filter_documents_by_format tests/track/track_test.py::DocumentCorpusTests::test_filter_documents_by_format_and_indices tests/track/track_test.py::DocumentCorpusTests::test_filter_documents_by_indices
esrally/track/track.py esrally/track/loader.py
esrally/track/track.py esrally/track/loader.py