diff --git a/cumulusci/tasks/bulkdata/step.py b/cumulusci/tasks/bulkdata/step.py
index 4ae6c50cca..766c660438 100644
--- a/cumulusci/tasks/bulkdata/step.py
+++ b/cumulusci/tasks/bulkdata/step.py
@@ -463,6 +463,18 @@ def select_records(self, records):
         records, records_copy = tee(records)
         # Count total number of records to fetch using the copy
         total_num_records = sum(1 for _ in records_copy)
+
+        # In the case that records are zero, return success
+        if total_num_records == 0:
+            self.logger.info(f"No records present for {self.sobject}")
+            self.job_result = DataOperationJobResult(
+                status=DataOperationStatus.SUCCESS,
+                job_errors=[],
+                records_processed=0,
+                total_row_errors=0,
+            )
+            return
+
         limit_clause = self._determine_limit_clause(total_num_records=total_num_records)
 
         # Generate and execute SOQL query
@@ -882,6 +894,17 @@ def select_records(self, records):
         # Count total number of records to fetch using the copy
         total_num_records = sum(1 for _ in records_copy)
 
+        # In the case that records are zero, return success
+        self.logger.info(f"No records present for {self.sobject}")
+        if total_num_records == 0:
+            self.job_result = DataOperationJobResult(
+                status=DataOperationStatus.SUCCESS,
+                job_errors=[],
+                records_processed=0,
+                total_row_errors=0,
+            )
+            return
+
         # Set LIMIT condition
         limit_clause = self._determine_limit_clause(total_num_records)
 
diff --git a/cumulusci/tasks/bulkdata/tests/test_step.py b/cumulusci/tasks/bulkdata/tests/test_step.py
index 3887b270f3..25a8362a54 100644
--- a/cumulusci/tasks/bulkdata/tests/test_step.py
+++ b/cumulusci/tasks/bulkdata/tests/test_step.py
@@ -591,6 +591,47 @@ def test_select_records_standard_strategy_success(self, download_mock):
             == 3
         )
 
+    @mock.patch("cumulusci.tasks.bulkdata.step.download_file")
+    def test_select_records_zero_load_records(self, download_mock):
+        # Set up mock context and BulkApiDmlOperation
+        context = mock.Mock()
+        step = BulkApiDmlOperation(
+            sobject="Contact",
+            operation=DataOperationType.QUERY,
+            api_options={"batch_size": 10, "update_key": "LastName"},
+            context=context,
+            fields=["LastName"],
+            selection_strategy=SelectStrategy.STANDARD,
+            content_type="JSON",
+        )
+
+        # Mock Bulk API responses
+        step.bulk.endpoint = "https://test"
+        step.bulk.create_query_job.return_value = "JOB"
+        step.bulk.query.return_value = "BATCH"
+        step.bulk.get_query_batch_result_ids.return_value = ["RESULT"]
+
+        # Mock the downloaded CSV content with a single record
+        download_mock.return_value = io.StringIO('[{"Id":"003000000000001"}]')
+
+        # Mock the _wait_for_job method to simulate a successful job
+        step._wait_for_job = mock.Mock()
+        step._wait_for_job.return_value = DataOperationJobResult(
+            DataOperationStatus.SUCCESS, [], 0, 0
+        )
+
+        # Prepare input records
+        records = iter([])
+
+        # Execute the select_records operation
+        step.start()
+        step.select_records(records)
+        step.end()
+
+        # Get the results and assert their properties
+        results = list(step.get_results())
+        assert len(results) == 0  # Expect 0 results (no records to process)
+
     @mock.patch("cumulusci.tasks.bulkdata.step.download_file")
     def test_select_records_standard_strategy_failure__no_records(self, download_mock):
         # Set up mock context and BulkApiDmlOperation
@@ -1927,6 +1968,45 @@ def test_select_records_standard_strategy_success(self):
             == 3
         )
 
+    @responses.activate
+    def test_select_records_zero_load_records(self):
+        mock_describe_calls()
+        task = _make_task(
+            LoadData,
+            {
+                "options": {
+                    "database_url": "sqlite:///test.db",
+                    "mapping": "mapping.yml",
+                }
+            },
+        )
+        task.project_config.project__package__api_version = CURRENT_SF_API_VERSION
+        task._init_task()
+
+        step = RestApiDmlOperation(
+            sobject="Contact",
+            operation=DataOperationType.UPSERT,
+            api_options={"batch_size": 10, "update_key": "LastName"},
+            context=task,
+            fields=["LastName"],
+            selection_strategy=SelectStrategy.STANDARD,
+        )
+
+        results = {
+            "records": [],
+            "done": True,
+        }
+        step.sf.restful = mock.Mock()
+        step.sf.restful.return_value = results
+        records = iter([])
+        step.start()
+        step.select_records(records)
+        step.end()
+
+        # Get the results and assert their properties
+        results = list(step.get_results())
+        assert len(results) == 0  # Expect 0 results (matching the input records count)
+
     @responses.activate
     def test_select_records_standard_strategy_success_pagination(self):
         mock_describe_calls()