Repository: SFDO-Tooling/CumulusCI
Pull request #3099 (2022-02-28 17:23:56)

Native Upserts for LoadData
Adds the ability to do Upserts during Data Loading.

The place to start is probably with the [documentation](https://github.com/SFDO-Tooling/CumulusCI/pull/3099/files#diff-194a6eb60a3032894acafae20fb216b5c4c69d6eea231ed18bab3e2186abdd15).



Patch:
diff --git a/cumulusci.yml b/cumulusci.yml
index 812e8394c..8e6cabbda 100644
--- a/cumulusci.yml
+++ b/cumulusci.yml
@@ -40,6 +40,12 @@ tasks:
                 - cumulusci/robotframework/Salesforce.robot
                 - cumulusci/robotframework/pageobjects/BasePageObjects.py
 
+    test_upserts:
+        description: Test the upsert feature
+        class_path: cumulusci.tasks.bulkdata.LoadData
+        options:
+            sql_path: datasets/upsert/upsert/upsert_before_data.sql
+            mapping: datasets/upsert/upsert/upsert_mapping.yml
 flows:
     qa_org:
         steps:
diff --git a/cumulusci/tasks/bulkdata/load.py b/cumulusci/tasks/bulkdata/load.py
index 020343bd3..6b2c6a568 100644
--- a/cumulusci/tasks/bulkdata/load.py
+++ b/cumulusci/tasks/bulkdata/load.py
@@ -173,10 +173,13 @@ class LoadData(SqlAlchemyMixin, BaseSalesforceApiTask):
 
         query = self._query_db(mapping)
         bulk_mode = mapping.bulk_mode or self.bulk_mode or "Parallel"
+        api_options = {"batch_size": mapping.batch_size, "bulk_mode": bulk_mode}
+        if mapping.update_key:
+            api_options["update_key"] = mapping.update_key
         step = get_dml_operation(
             sobject=mapping.sf_object,
             operation=mapping.action,
-            api_options={"batch_size": mapping.batch_size, "bulk_mode": bulk_mode},
+            api_options=api_options,
             context=self,
             fields=mapping.get_load_field_list(),
             api=mapping.api,
diff --git a/cumulusci/tasks/bulkdata/mapping_parser.py b/cumulusci/tasks/bulkdata/mapping_parser.py
index 2fae820cd..d9e51b17b 100644
--- a/cumulusci/tasks/bulkdata/mapping_parser.py
+++ b/cumulusci/tasks/bulkdata/mapping_parser.py
@@ -1,3 +1,4 @@
+import typing as T
 from datetime import date
 from enum import Enum
 from logging import getLogger
@@ -98,6 +99,7 @@ class MappingStep(CCIDictModel):
     ] = None  # default should come from task options
     anchor_date: Optional[Union[str, date]] = None
     soql_filter: Optional[str] = None  # soql_filter property
+    update_key: str = None  # only for upserts
 
     @validator("bulk_mode", "api", "action", pre=True)
     def case_normalize(cls, val):
@@ -253,35 +255,56 @@ class MappingStep(CCIDictModel):
             lookup.name = name
         return v
 
+    @root_validator
+    @classmethod
+    def validate_external_id_and_upsert(cls, v):
+        # TODO: should also validate that update_key is in fields list.
+        has_update_key = bool(v.get("update_key"))
+        is_upsert = v["action"] == DataOperationType.UPSERT
+        assert (
+            has_update_key == is_upsert
+        ), "Action 'upsert' and option 'update_key' must always be used together."
+        return v
+
     @staticmethod
     def _is_injectable(element: str) -> bool:
         return element.count("__") == 1
 
-    def _get_permission_type(self, operation: DataOperationType) -> str:
+    def _get_required_permission_types(
+        self, operation: DataOperationType
+    ) -> T.Tuple[str]:
+        """Return a tuple of the permssion types required to execute an operaation"""
         if operation is DataOperationType.QUERY:
-            return "queryable"
+            return ("queryable",)
         if (
             operation is DataOperationType.INSERT
             and self.action is DataOperationType.UPDATE
         ):
-            return "updateable"
+            return ("updateable",)
+        if (
+            operation is DataOperationType.UPSERT
+            or self.action is DataOperationType.UPSERT
+        ):
+            return ("updateable", "createable")
 
-        return "createable"
+        return ("createable",)
 
     def _check_object_permission(
         self, global_describe: Mapping, sobject: str, operation: DataOperationType
     ):
         assert sobject in global_describe
-        perm = self._get_permission_type(operation)
-        return global_describe[sobject][perm]
+        perms = self._get_required_permission_types(operation)
+        return all(global_describe[sobject][perm] for perm in perms)
 
     def _check_field_permission(
         self, describe: Mapping, field: str, operation: DataOperationType
     ):
-        perm = self._get_permission_type(operation)
+        perms = self._get_required_permission_types(operation)
         # Fields don't have "queryable" permission.
-        return field in describe and (
+        return field in describe and all(
+            # To discuss: is this different than `describe[field].get(perm, True)`
             describe[field].get(perm) if perm in describe[field] else True
+            for perm in perms
         )
 
     def _validate_field_dict(
diff --git a/cumulusci/tasks/bulkdata/step.py b/cumulusci/tasks/bulkdata/step.py
index 7b5c1509d..d5df2a0fe 100644
--- a/cumulusci/tasks/bulkdata/step.py
+++ b/cumulusci/tasks/bulkdata/step.py
@@ -30,6 +30,7 @@ class DataOperationType(Enum):
     DELETE = "delete"
     HARD_DELETE = "hardDelete"
     QUERY = "query"
+    UPSERT = "upsert"
 
 
 class DataApi(Enum):
@@ -346,6 +347,7 @@ class BulkApiDmlOperation(BaseDmlOperation, BulkJobMixin):
             self.operation.value,
             contentType="CSV",
             concurrency=self.api_options.get("bulk_mode", "Parallel"),
+            external_id_name=self.api_options.get("update_key"),
         )
 
     def end(self):
@@ -489,14 +491,20 @@ class RestApiDmlOperation(BaseDmlOperation):
             DataOperationType.INSERT: "POST",
             DataOperationType.UPDATE: "PATCH",
             DataOperationType.DELETE: "DELETE",
+            DataOperationType.UPSERT: "PATCH",
         }[self.operation]
 
+        update_key = self.api_options.get("update_key")
         for chunk in iterate_in_chunks(self.api_options.get("batch_size"), records):
             if self.operation is DataOperationType.DELETE:
                 url_string = "?ids=" + ",".join(_convert(rec)["Id"] for rec in chunk)
                 json = None
             else:
-                url_string = ""
+                if update_key:
+                    assert self.operation == DataOperationType.UPSERT
+                    url_string = f"/{self.sobject}/{update_key}"
+                else:
+                    url_string = ""
                 json = {"allOrNone": False, "records": [_convert(rec) for rec in chunk]}
 
             self.results.extend(
@@ -594,6 +602,8 @@ def get_dml_operation(
     between REST and Bulk APIs based upon volume (Bulk used at volumes over 2000 records,
     or if the operation is HARD_DELETE, which is only available for Bulk)."""
 
+    context.logger.debug(f"Creating {operation} Operation for {sobject} using {api}")
+
     # REST Collections requires 42.0.
     api_version = float(context.sf.sf_version)
     if api_version < 42.0 and api is not DataApi.BULK:
diff --git a/datasets/upsert/upsert_before_data.sql b/datasets/upsert/upsert_before_data.sql
new file mode 100644
index 000000000..9d65829eb
--- /dev/null
+++ b/datasets/upsert/upsert_before_data.sql
@@ -0,0 +1,31 @@
+BEGIN TRANSACTION;
+CREATE TABLE "Account" (
+	id INTEGER NOT NULL, 
+	"Name" VARCHAR(255), 
+	PRIMARY KEY (id)
+);
+INSERT INTO "Account" VALUES(1,'Sitwell-Bluth');
+CREATE TABLE "Contact" (
+	id INTEGER NOT NULL, 
+	"FirstName" VARCHAR(255), 
+	"LastName" VARCHAR(255), 
+	"Email" VARCHAR(255), 
+	PRIMARY KEY (id)
+);
+INSERT INTO "Contact" VALUES(1,'Lindsay','Sitwell','lindsay.bluth@example.com');
+INSERT INTO "Contact" VALUES(2,'Audrey','Cain','audrey.cain@example.com');
+INSERT INTO "Contact" VALUES(3,'Micheal','Bernard','michael.bernard@example.com');
+INSERT INTO "Contact" VALUES(4,'Chloe','Myers','Chloe.Myers@example.com');
+INSERT INTO "Contact" VALUES(5,'Rose','Larson','Rose.Larson@example.com');
+INSERT INTO "Contact" VALUES(6,'Brent','Ali','Brent.Ali@example.com');
+INSERT INTO "Contact" VALUES(7,'Julia','Townsend','Julia.Townsend@example.com');
+INSERT INTO "Contact" VALUES(8,'Benjamin','Cunningham','Benjamin.Cunningham@example.com');
+INSERT INTO "Contact" VALUES(9,'Christy','Stanton','Christy.Stanton@example.com');
+INSERT INTO "Contact" VALUES(10,'Sabrina','Roberson','Sabrina.Roberson@example.com');
+INSERT INTO "Contact" VALUES(11,'Michael','Bluth','Michael.Bluth@example.com');
+INSERT INTO "Contact" VALUES(12,'Javier','Banks','Javier.Banks@example.com');
+INSERT INTO "Contact" VALUES(13,'GOB','Bluth','GOB.Bluth@example.com');
+INSERT INTO "Contact" VALUES(14,'Kaitlyn','Rubio','Kaitlyn.Rubio@example.com');
+INSERT INTO "Contact" VALUES(15,'Jerry','Eaton','Jerry.Eaton@example.com');
+INSERT INTO "Contact" VALUES(16,'Gabrielle','Vargas','Gabrielle.Vargas@example.com');
+COMMIT;
diff --git a/datasets/upsert/upsert_example_2.sql b/datasets/upsert/upsert_example_2.sql
new file mode 100644
index 000000000..e5d1e318d
--- /dev/null
+++ b/datasets/upsert/upsert_example_2.sql
@@ -0,0 +1,17 @@
+BEGIN TRANSACTION;
+CREATE TABLE "Account" (
+	id INTEGER NOT NULL, 
+	"Name" VARCHAR(255), 
+	PRIMARY KEY (id)
+);
+CREATE TABLE "Contact" (
+	id INTEGER NOT NULL, 
+	"FirstName" VARCHAR(255), 
+	"LastName" VARCHAR(255), 
+	"Email" VARCHAR(255),
+	PRIMARY KEY (id)
+);
+INSERT INTO "Contact" VALUES(1,'Nichael','Bluth','Michael.Bluth@example.com');
+INSERT INTO "Contact" VALUES(2,'George Oscar','Bluth','GOB.Bluth@example.com');
+INSERT INTO "Contact" VALUES(3,'Lindsay','Bluth','lindsay.bluth@example.com');
+COMMIT;
\ No newline at end of file
diff --git a/datasets/upsert/upsert_mapping_bulk.yml b/datasets/upsert/upsert_mapping_bulk.yml
new file mode 100644
index 000000000..030dc380e
--- /dev/null
+++ b/datasets/upsert/upsert_mapping_bulk.yml
@@ -0,0 +1,14 @@
+Insert Accounts:
+    api: bulk
+    sf_object: Account
+    fields:
+        - Name
+Insert Contacts:
+    api: bulk
+    sf_object: Contact
+    action: upsert
+    update_key: Email
+    fields:
+        - FirstName
+        - LastName
+        - Email
diff --git a/datasets/upsert/upsert_mapping_rest.yml b/datasets/upsert/upsert_mapping_rest.yml
new file mode 100644
index 000000000..47b268af2
--- /dev/null
+++ b/datasets/upsert/upsert_mapping_rest.yml
@@ -0,0 +1,14 @@
+Insert Accounts:
+    api: rest
+    sf_object: Account
+    fields:
+        - Name
+Insert Contacts:
+    api: rest
+    sf_object: Contact
+    action: upsert
+    update_key: Email
+    fields:
+        - FirstName
+        - LastName
+        - Email
diff --git a/datasets/upsert/upsert_mapping_rest__external_id.yml b/datasets/upsert/upsert_mapping_rest__external_id.yml
new file mode 100644
index 000000000..47b268af2
--- /dev/null
+++ b/datasets/upsert/upsert_mapping_rest__external_id.yml
@@ -0,0 +1,14 @@
+Insert Accounts:
+    api: rest
+    sf_object: Account
+    fields:
+        - Name
+Insert Contacts:
+    api: rest
+    sf_object: Contact
+    action: upsert
+    update_key: Email
+    fields:
+        - FirstName
+        - LastName
+        - Email
diff --git a/docs/data.rst b/docs/data.rst
index bc266a3aa..87be08d16 100644
--- a/docs/data.rst
+++ b/docs/data.rst
@@ -111,6 +111,37 @@ which is subject to its own limits, including restrictions on total processing t
 Bulk API batches are automatically chunked further into transactions by the platform,
 and the transaction size cannot be controlled.
 
+Upserts
+-------
+The definition of "upsert" is an operation which creates new records and updates 
+existing records depending on a field (the update key) which determines whether
+the input row and the existing row are "the same".
+
+
+You can do ID-based, 
+`idLookup-based <https://developer.salesforce.com/docs/atlas.en-us.204.0.object_reference.meta/object_reference/access_for_fields.htm#access_lookup>`_ 
+and `external ID <https://help.salesforce.com/s/articleView?id=sf.faq_import_general_what_is_an_external.htm&type=5>`_-based
+`upserts <https://developer.salesforce.com/docs/atlas.en-us.apexcode.meta/apexcode/langCon_apex_dml_examples_upsert.htm>`_ 
+and updates by specifying additional settings in a mapping step.
+
+Objects inserted through an upsert are not able to be referred to through lookups uploaded in the same
+```load_data`` task.
+
+.. code-block:: yaml
+
+    Insert Accounts:
+        sf_object: Account
+        action: upsert
+        update_key: Extid__c
+        fields:
+            - Name
+            - Extid__c
+
+Whenever ``update_key`` is supplied, the action must be ``upsert`` and
+vice versa.
+
+
+
 Database Mapping
 ----------------
 


Test patch:
diff --git a/cumulusci/tasks/bulkdata/tests/test_mapping_parser.py b/cumulusci/tasks/bulkdata/tests/test_mapping_parser.py
index f8de541d9..b25006557 100644
--- a/cumulusci/tasks/bulkdata/tests/test_mapping_parser.py
+++ b/cumulusci/tasks/bulkdata/tests/test_mapping_parser.py
@@ -219,17 +219,34 @@ class TestMappingParser:
         assert not MappingStep._is_injectable("npsp__Test__c")
         assert not MappingStep._is_injectable("Account")
 
-    def test_get_permission_type(self):
+    def test_get_permission_types(self):
         ms = MappingStep(
             sf_object="Account", fields=["Name"], action=DataOperationType.INSERT
         )
-        assert ms._get_permission_type(DataOperationType.INSERT) == "createable"
-        assert ms._get_permission_type(DataOperationType.QUERY) == "queryable"
+        assert ms._get_required_permission_types(DataOperationType.INSERT) == (
+            "createable",
+        )
+        assert ms._get_required_permission_types(DataOperationType.QUERY) == (
+            "queryable",
+        )
 
         ms = MappingStep(
             sf_object="Account", fields=["Name"], action=DataOperationType.UPDATE
         )
-        assert ms._get_permission_type(DataOperationType.INSERT) == "updateable"
+        assert ms._get_required_permission_types(DataOperationType.INSERT) == (
+            "updateable",
+        )
+
+        ms = MappingStep(
+            sf_object="Account",
+            fields=["Name"],
+            action=DataOperationType.UPSERT,
+            update_key="Extid__c",
+        )
+        assert ms._get_required_permission_types(DataOperationType.UPSERT) == (
+            "updateable",
+            "createable",
+        )
 
     def test_check_field_permission(self):
         ms = MappingStep(
diff --git a/cumulusci/tasks/bulkdata/tests/test_step.py b/cumulusci/tasks/bulkdata/tests/test_step.py
index b24bdc114..374bae4b6 100644
--- a/cumulusci/tasks/bulkdata/tests/test_step.py
+++ b/cumulusci/tasks/bulkdata/tests/test_step.py
@@ -391,7 +391,11 @@ class TestBulkApiDmlOperation(unittest.TestCase):
         step.start()
 
         context.bulk.create_job.assert_called_once_with(
-            "Contact", "insert", contentType="CSV", concurrency="Parallel"
+            "Contact",
+            "insert",
+            contentType="CSV",
+            concurrency="Parallel",
+            external_id_name=None,
         )
         assert step.job_id == "JOB"
 
@@ -462,7 +466,11 @@ class TestBulkApiDmlOperation(unittest.TestCase):
             pass
 
         context.bulk.create_job.assert_called_once_with(
-            "Contact", "insert", contentType="CSV", concurrency="Parallel"
+            "Contact",
+            "insert",
+            contentType="CSV",
+            concurrency="Parallel",
+            external_id_name=None,
         )
         assert step.job_id == "JOB"
 
diff --git a/cumulusci/tasks/bulkdata/tests/test_upsert.py b/cumulusci/tasks/bulkdata/tests/test_upsert.py
new file mode 100644
index 000000000..2c2f4010a
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/tests/test_upsert.py
@@ -0,0 +1,378 @@
+import re
+from unittest import mock
+
+import pytest
+import responses
+
+from cumulusci.tasks.bulkdata import LoadData
+from cumulusci.tasks.bulkdata.step import DataApi, DataOperationStatus
+from cumulusci.tests.util import mock_describe_calls
+
+CURRENT_SF_API_VERSION = "52.0"  # match cumulusci.yml until it updates
+
+
+class TestUpsert:
+    # bulk API not supported by VCR yet
+    @pytest.mark.needs_org()
+    def test_upsert_external_id_field(
+        self, create_task, cumulusci_test_repo_root, sf, delete_data_from_org
+    ):
+        delete_data_from_org(["Entitlement", "Opportunity", "Contact", "Account"])
+        self._test_two_upserts_and_check_results(
+            "bulk", create_task, cumulusci_test_repo_root, sf
+        )
+
+    def _test_two_upserts_and_check_results(
+        self, api, create_task, cumulusci_test_repo_root, sf
+    ):
+        task = create_task(
+            LoadData,
+            {
+                "sql_path": cumulusci_test_repo_root
+                / "datasets/upsert/upsert_before_data.sql",
+                "mapping": cumulusci_test_repo_root
+                / f"datasets/upsert/upsert_mapping_{api}.yml",
+                "ignore_row_errors": True,
+                "set_recently_viewed": False,
+            },
+        )
+        with mock.patch.object(task.logger, "info"):
+            task()
+            result = task.return_values
+            assert result == {
+                "step_results": {
+                    "Insert Accounts": {
+                        "sobject": "Account",
+                        "record_type": None,
+                        "status": DataOperationStatus.SUCCESS,
+                        "job_errors": [],
+                        "records_processed": 1,
+                        "total_row_errors": 0,
+                    },
+                    "Insert Contacts": {
+                        "sobject": "Contact",
+                        "record_type": None,
+                        "status": DataOperationStatus.SUCCESS,
+                        "job_errors": [],
+                        "records_processed": 16,
+                        "total_row_errors": 0,
+                    },
+                }
+            }, result
+            accounts = sf.query("select Name from Account")
+            accounts = {account["Name"] for account in accounts["records"]}
+            assert "Sitwell-Bluth" in accounts
+            contacts = sf.query("select FirstName from Contact")
+            contacts = {contact["FirstName"] for contact in contacts["records"]}
+            assert "Nichael" not in contacts
+            assert "George Oscar" not in contacts
+            assert "UPSERT" in str(task.logger.info.mock_calls)
+
+        with mock.patch.object(task.logger, "info"):
+            task = create_task(
+                LoadData,
+                {
+                    "sql_path": cumulusci_test_repo_root
+                    / "datasets/upsert/upsert_example_2.sql",
+                    "mapping": cumulusci_test_repo_root
+                    / f"datasets/upsert/upsert_mapping_{api}.yml",
+                    "ignore_row_errors": True,
+                    "set_recently_viewed": False,
+                },
+            )
+            task()
+            result = task.return_values
+            contacts = sf.query("select FirstName from Contact")
+            contacts = {contact["FirstName"] for contact in contacts["records"]}
+
+            assert "Nichael" in contacts
+            assert "George Oscar" in contacts
+            assert "UPSERT" in str(task.logger.info.mock_calls)
+
+    @pytest.mark.needs_org()
+    def test_upsert__rest(
+        self,
+        create_task,
+        cumulusci_test_repo_root,
+        run_code_without_recording,
+        delete_data_from_org,
+        sf,
+    ):
+
+        run_code_without_recording(
+            lambda: delete_data_from_org(
+                ["Entitlement", "Opportunity", "Contact", "Account"]
+            )
+        )
+        self._test_two_upserts_and_check_results(
+            "rest",
+            create_task,
+            cumulusci_test_repo_root,
+            sf,
+        )
+
+    @responses.activate
+    def test_upsert_rest__faked(
+        self, create_task, cumulusci_test_repo_root, org_config
+    ):
+        domain = org_config.get_domain()
+        ver = CURRENT_SF_API_VERSION
+        responses.add(
+            method="PATCH",
+            url=f"https://{domain}/services/data/v{ver}/composite/sobjects/Contact/Email",
+            status=200,
+            json=[
+                {
+                    "id": "003P000001Y5exdIAB",
+                    "success": True,
+                    "errors": [],
+                    "created": True,
+                },
+                {
+                    "id": "003P000001Y5exXIAR",
+                    "success": True,
+                    "errors": [],
+                    "created": False,
+                },
+                {
+                    "id": "003P000001Y5exYIAR",
+                    "success": True,
+                    "errors": [],
+                    "created": False,
+                },
+            ],
+        )
+        mock_describe_calls(domain=domain, version=CURRENT_SF_API_VERSION)
+        task = create_task(
+            LoadData,
+            {
+                "sql_path": cumulusci_test_repo_root
+                / "datasets/upsert/upsert_example_2.sql",
+                "mapping": cumulusci_test_repo_root
+                / "datasets/upsert/upsert_mapping_rest__external_id.yml",
+                "ignore_row_errors": True,
+                "set_recently_viewed": False,
+            },
+        )
+        task._update_credentials = mock.Mock()
+        with mock.patch.object(task.logger, "debug"):
+            rc = task()
+            assert rc == {
+                "step_results": {
+                    "Insert Accounts": {
+                        "sobject": "Account",
+                        "record_type": None,
+                        "status": DataOperationStatus.SUCCESS,
+                        "job_errors": [],
+                        "records_processed": 0,
+                        "total_row_errors": 0,
+                    },
+                    "Insert Contacts": {
+                        "sobject": "Contact",
+                        "record_type": None,
+                        "status": DataOperationStatus.SUCCESS,
+                        "job_errors": [],
+                        "records_processed": 3,
+                        "total_row_errors": 0,
+                    },
+                }
+            }, rc
+
+            relevant_debug_statement = look_for_operation_creation_debug_statement(
+                task.logger.debug.mock_calls
+            )
+            assert relevant_debug_statement == str(
+                DataApi.REST
+            ), relevant_debug_statement
+
+    @responses.activate
+    def test_upsert__fake_bulk(self, create_task, cumulusci_test_repo_root, org_config):
+        domain = org_config.get_domain()
+        mock_describe_calls(domain=domain, version=CURRENT_SF_API_VERSION)
+        responses.add(
+            method="POST",
+            url=f"https://{domain}/services/async/{CURRENT_SF_API_VERSION}/job",
+            status=200,
+            body=f"""<?xml version="1.0" encoding="UTF-8"?>
+            <jobInfo xmlns="http://www.force.com/2009/06/asyncapi/dataload">
+        <id>750P0000005mX3LIAU</id>
+        <operation>insert</operation>
+        <object>Account</object>
+        <createdById>005P0000009ajX7IAI</createdById>
+        <createdDate>2022-03-02T04:14:43.000Z</createdDate>
+        <systemModstamp>2022-03-02T04:14:43.000Z</systemModstamp>
+        <state>Open</state>
+        <concurrencyMode>Parallel</concurrencyMode>
+        <contentType>CSV</contentType>
+        <numberBatchesQueued>0</numberBatchesQueued>
+        <numberBatchesInProgress>0</numberBatchesInProgress>
+        <numberBatchesCompleted>0</numberBatchesCompleted>
+        <numberBatchesFailed>0</numberBatchesFailed>
+        <numberBatchesTotal>0</numberBatchesTotal>
+        <numberRecordsProcessed>0</numberRecordsProcessed>
+        <numberRetries>0</numberRetries>
+        <apiVersion>v{CURRENT_SF_API_VERSION}</apiVersion>
+        <numberRecordsFailed>0</numberRecordsFailed>
+        <totalProcessingTime>0</totalProcessingTime>
+        <apiActiveProcessingTime>0</apiActiveProcessingTime>
+        <apexProcessingTime>0</apexProcessingTime>
+        </jobInfo>""",
+        )
+        responses.add(
+            method="POST",
+            url=f"https://{domain}/services/async/{CURRENT_SF_API_VERSION}/job/750P0000005mX3LIAU",
+            status=200,
+            body=f""""<?xml version="1.0" encoding="UTF-8"?>
+            <jobInfo xmlns="http://www.force.com/2009/06/asyncapi/dataload">
+            <id>750P0000005mX3LIAU</id>
+            <operation>insert</operation>
+            <object>Account</object>
+            <createdById>005P0000009ajX7IAI</createdById>
+            <createdDate>2022-03-02T04:14:43.000Z</createdDate>
+            <systemModstamp>2022-03-02T04:14:43.000Z</systemModstamp>
+            <state>Closed</state>
+            <concurrencyMode>Parallel</concurrencyMode>
+            <contentType>CSV</contentType>
+            <numberBatchesQueued>0</numberBatchesQueued>
+            <numberBatchesInProgress>0</numberBatchesInProgress>
+            <numberBatchesCompleted>0</numberBatchesCompleted>
+            <numberBatchesFailed>0</numberBatchesFailed>
+            <numberBatchesTotal>0</numberBatchesTotal>
+            <numberRecordsProcessed>0</numberRecordsProcessed>
+            <numberRetries>0</numberRetries>
+            <apiVersion>v{CURRENT_SF_API_VERSION}.0</apiVersion>
+            <numberRecordsFailed>0</numberRecordsFailed>
+            <totalProcessingTime>0</totalProcessingTime>
+            <apiActiveProcessingTime>0</apiActiveProcessingTime>
+            <apexProcessingTime>0</apexProcessingTime>
+            </jobInfo>""",
+        )
+
+        responses.add(
+            method="GET",
+            url=f"https://{domain}/services/async/{CURRENT_SF_API_VERSION}/job/750P0000005mX3LIAU",
+            status=200,
+            body="""<?xml version="1.0" encoding="UTF-8"?>
+            <jobInfo xmlns="http://www.force.com/2009/06/asyncapi/dataload">
+
+            <id>750P0000005mX3LIAU</id>
+            <operation>insert</operation>
+            <object>Account</object>
+            <createdById>005P0000009ajX7IAI</createdById>
+            <createdDate>2022-03-02T04:14:43.000Z</createdDate>
+            <systemModstamp>2022-03-02T04:14:43.000Z</systemModstamp>
+            <state>Closed</state>
+            <concurrencyMode>Parallel</concurrencyMode>
+            <contentType>CSV</contentType>
+            <numberBatchesQueued>0</numberBatchesQueued>
+            <numberBatchesInProgress>0</numberBatchesInProgress>
+            <numberBatchesCompleted>0</numberBatchesCompleted>
+            <numberBatchesFailed>0</numberBatchesFailed>
+            <numberBatchesTotal>0</numberBatchesTotal>
+            <numberRecordsProcessed>4</numberRecordsProcessed>
+            <numberRetries>0</numberRetries>
+            <apiVersion>vxx.0</apiVersion>
+            <numberRecordsFailed>0</numberRecordsFailed>
+            <totalProcessingTime>0</totalProcessingTime>
+            <apiActiveProcessingTime>0</apiActiveProcessingTime>
+            <apexProcessingTime>0</apexProcessingTime>
+            </jobInfo>""",
+        )
+        responses.add(
+            method="GET",
+            url=f"https://{domain}/services/async/{CURRENT_SF_API_VERSION}/job/750P0000005mX3LIAU/batch",
+            status=200,
+            body="""<?xml version="1.0" encoding="UTF-8"?>
+            <batchInfoList xmlns="http://www.force.com/2009/06/asyncapi/dataload"/>""",
+        )
+
+        responses.add(
+            method="POST",
+            url=f"https://{domain}/services/async/{CURRENT_SF_API_VERSION}/job/750P0000005mX3LIAU/batch",
+            status=200,
+            body="""<?xml version="1.0" encoding="UTF-8"?>
+            <batchInfo xmlns="http://www.force.com/2009/06/asyncapi/dataload">
+
+        <id>751P0000006fIUAIA2</id>
+        <jobId>750P0000005mX2wIAE</jobId>
+        <state>Queued</state>
+        <createdDate>2022-03-02T04:14:44.000Z</createdDate>
+        <systemModstamp>2022-03-02T04:14:44.000Z</systemModstamp>
+        <numberRecordsProcessed>4</numberRecordsProcessed>
+        <numberRecordsFailed>0</numberRecordsFailed>
+        <totalProcessingTime>0</totalProcessingTime>
+        <apiActiveProcessingTime>0</apiActiveProcessingTime>
+        <apexProcessingTime>0</apexProcessingTime>
+        </batchInfo>""",
+        )
+
+        responses.add(
+            method="GET",
+            url=f"https://{domain}/services/async/{CURRENT_SF_API_VERSION}/job/750P0000005mX3LIAU/batch/751P0000006fIUAIA2/result",
+            status=200,
+            body="""""Id","Success","Created","Error"
+        "003P000001Y5yvdIAB","true","false",""
+        "003P000001Y5ywMIAR","true","false",""
+        "003P000001Y5ywNIAR","true","false",""
+        "003P000001Y5ywOIAR","true","false","" """,
+        )
+
+        task = create_task(
+            LoadData,
+            {
+                "sql_path": cumulusci_test_repo_root
+                / "datasets/upsert/upsert_example_2.sql",
+                "mapping": cumulusci_test_repo_root
+                / "datasets/upsert/upsert_mapping_bulk.yml",
+                "ignore_row_errors": True,
+                "set_recently_viewed": False,
+            },
+        )
+        task._update_credentials = mock.Mock()
+
+        with mock.patch.object(task.logger, "debug"):
+            ret = task()
+            assert ret == {
+                "step_results": {
+                    "Insert Accounts": {
+                        "sobject": "Account",
+                        "record_type": None,
+                        "status": DataOperationStatus.SUCCESS,
+                        "job_errors": [],
+                        "records_processed": 0,
+                        "total_row_errors": 0,
+                    },
+                    "Insert Contacts": {
+                        "sobject": "Contact",
+                        "record_type": None,
+                        "status": DataOperationStatus.SUCCESS,
+                        "job_errors": [],
+                        "records_processed": 0,  # change here and above to 4 to match data
+                        "total_row_errors": 0,
+                    },
+                }
+            }, ret
+            relevant_debug_statement = look_for_operation_creation_debug_statement(
+                task.logger.debug.mock_calls
+            )
+            assert relevant_debug_statement == str(
+                DataApi.BULK
+            ), relevant_debug_statement
+
+
+def look_for_operation_creation_debug_statement(mock_calls):
+    relevant_debug_statements = (
+        look_for_operation_creation_debug_statement_for_string(call.args[0])
+        for call in mock_calls
+    )
+
+    return next(stmt for stmt in relevant_debug_statements if stmt)
+
+
+DEBUG_MATCHER = re.compile("Creating (.*) Operation for (.*) using (.*)")
+
+
+def look_for_operation_creation_debug_statement_for_string(s):
+    match = DEBUG_MATCHER.match(s)
+    if match:
+        return match[3]
diff --git a/cumulusci/tests/util.py b/cumulusci/tests/util.py
index d6f468411..641790b7d 100644
--- a/cumulusci/tests/util.py
+++ b/cumulusci/tests/util.py
@@ -198,7 +198,7 @@ def read_mock(name: str):
         return f.read()
 
 
-def mock_describe_calls(domain="example.com", version="54.0"):
+def mock_describe_calls(domain="example.com", version=CURRENT_SF_API_VERSION):
     def mock_sobject_describe(name: str):
         responses.add(
             method="GET",