Repository: SFDO-Tooling/CumulusCI
Pull request #3324 (2022-08-17 21:15:21)

Generate mappings and next generation datasets
Infrastructure required to extract from orgs without creating a mapping file (by auto-generating a mapping file internally).

This PR is essentially a lot of algorithms for looking at a real org Schema + Extract Declarations and generating two mapping files, one for Extracting data and one for Loading data.

No user-interfaces exist yet for these new features.

A new abstraction called a Dataset will go in another PR.


Patch:
diff --git a/cumulusci/salesforce_api/filterable_objects.py b/cumulusci/salesforce_api/filterable_objects.py
index 5691630ed..cfd64829d 100644
--- a/cumulusci/salesforce_api/filterable_objects.py
+++ b/cumulusci/salesforce_api/filterable_objects.py
@@ -7,6 +7,8 @@ NOT_COUNTABLE = (
     "IdeaComment",  # you must filter using the following syntax: CommunityId = [single ID],
     "Vote",  # you must filter using the following syntax: ParentId = [single ID],
     "RecordActionHistory",  # Gack: 1133111327-118855 (1126216936)
+    "DashboardSnapshotResults",  # Gack: 1198622208-46932 (1126216936)
+    "RecordRecommendation",  # Implementation restriction: RecordRecommendation requires a filter on TargetSobjectType or TargetId or RecordRecommendationId
 )
 
 
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/calculate_dependencies.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/calculate_dependencies.py
index efd3f31f9..dc65b2c80 100644
--- a/cumulusci/tasks/bulkdata/extract_dataset_utils/calculate_dependencies.py
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/calculate_dependencies.py
@@ -9,15 +9,16 @@ from .synthesize_extract_declarations import (
 )
 
 
-class Dependency(T.NamedTuple):
+class SObjDependency(T.NamedTuple):
     table_name_from: str
     table_name_to: str
     field_name: str
+    priority: bool = False
 
 
 def _calculate_dependencies_for_declarations(
     decls: T.Sequence[SimplifiedExtractDeclaration], schema: Schema
-) -> T.Dict[str, T.List[Dependency]]:
+) -> T.Dict[str, T.List[SObjDependency]]:
     """Ensure that required lookups are fulfilled for list of SimplifiedExtractDeclarations
 
     Do this by adding their referent tables (in full) to the extract.
@@ -25,14 +26,14 @@ def _calculate_dependencies_for_declarations(
     """
     dependencies = {}
     for decl in decls:
-        new_dependencies = _calculate_dependencies_for_sobject(
+        new_dependencies = _collect_dependencies_for_sobject(
             decl.sf_object, decl.fields, schema, only_required_fields=False
         )
         dependencies.update(new_dependencies)
     return dependencies
 
 
-def _calculate_dependencies_for_sobject(
+def _collect_dependencies_for_sobject(
     source_sfobject: str,
     fields: T.List[str],
     schema: Schema,
@@ -57,7 +58,9 @@ def _calculate_dependencies_for_sobject(
             field_allowed = not (only_required_fields or field_disallowed)
             if field_info.requiredOnCreate or field_allowed:
                 dependencies.setdefault(source_sfobject, []).append(
-                    Dependency(source_sfobject, target, field_name)
+                    SObjDependency(
+                        source_sfobject, target, field_name, field_info.requiredOnCreate
+                    )
                 )
 
     return dependencies
@@ -91,7 +94,7 @@ def extend_declarations_to_include_referenced_tables(
                     target_table, required_fields, schema[target_table].fields
                 )
 
-                new_dependencies = _calculate_dependencies_for_sobject(
+                new_dependencies = _collect_dependencies_for_sobject(
                     target_table,
                     decls[target_table].fields,
                     schema,
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/synthesize_extract_declarations.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/synthesize_extract_declarations.py
index cda91c096..87653858a 100644
--- a/cumulusci/tasks/bulkdata/extract_dataset_utils/synthesize_extract_declarations.py
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/synthesize_extract_declarations.py
@@ -135,7 +135,7 @@ def _expand_group_sobject_declaration(decl: ExtractDeclaration, schema: Schema):
         assert 0, decl.group_type
 
     matching_objects = [
-        obj["name"] for obj in schema.sobjects if matches_obj(obj) and obj.count > 1
+        obj["name"] for obj in schema.sobjects if matches_obj(obj) and obj.count >= 1
     ]
     decls = [
         synthesize_declaration_for_sobject(obj, decl.fields, schema[obj].fields)
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/dependency_map.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/dependency_map.py
new file mode 100644
index 000000000..ae9c53022
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/dependency_map.py
@@ -0,0 +1,163 @@
+import typing as T
+from collections import defaultdict
+
+from cumulusci.tasks.bulkdata.extract_dataset_utils.calculate_dependencies import (
+    SObjDependency,
+)
+from cumulusci.utils.collections import OrderedSet, OrderedSetType
+
+StrDependencyMapping = T.Mapping[str, OrderedSetType[SObjDependency]]
+SObjectRuleDeclaration = (
+    "snowfakery.cci_mapping_files.declaration_parser.SObjectRuleDeclaration"
+)
+
+
+class DependencyMap:
+    """Index which tables depend on which other ones (through lookups)
+
+    To make lookups easy later.
+    """
+
+    # a dictionary allowing easy lookup of inferred/soft dependencies by parent table
+    soft_dependencies: StrDependencyMapping
+
+    # a dictionary allowing lookups by (tablename, fieldname) pairs
+    reference_fields: StrDependencyMapping
+
+    _sorted_tables = None  # cache for sorted table computation
+
+    def __init__(
+        self,
+        table_names: T.Iterable[str],
+        dependencies: T.Iterable[SObjDependency],
+    ):
+        self.dependencies = defaultdict(OrderedSet)
+        self.reference_fields = {}
+        self._map_references(dependencies)
+        self.table_names = tuple(table_names)
+
+    def _map_references(
+        self,
+        intertable_dependencies: T.Iterable[SObjDependency],
+    ):
+        for dep in intertable_dependencies:
+            table_deps = self.dependencies[dep.table_name_from]
+            table_deps.add(dep)
+            self.reference_fields[
+                (dep.table_name_from, dep.field_name)
+            ] = dep.table_name_to
+
+    def target_table_for(self, tablename: str, fieldname: str) -> T.Optional[str]:
+        return self.reference_fields.get((tablename, fieldname))
+
+    def get_dependency_order(self):
+        """Sort the dependencies to output tables in the right order."""
+        if not self._sorted_tables:
+            # SnowfakeryPersonAccounts: Add this back in when Snowfakery is integrated with this code.
+            # _remove_person_contact_id(self.dependencies)
+            self._sorted_tables = _sort_by_dependencies(
+                tuple(self.table_names), self.dependencies
+            )
+        return self._sorted_tables
+
+
+def _sort_by_dependencies(
+    table_names: T.Sequence[str],
+    dependencies: T.Mapping[str, OrderedSet],
+    priority: int = 0,
+):
+    "Sort tables by dependency relationships."
+    sorted_tables = []
+    dependencies = dict(dependencies)
+
+    while table_names:
+        remaining = len(table_names)
+        leaf_tables = [
+            table
+            for table in table_names
+            if _table_is_free(table, dependencies, sorted_tables, priority)
+        ]
+        sorted_tables.extend(leaf_tables)
+        table_names = [table for table in table_names if table not in sorted_tables]
+
+        # Nothing changed, so we need to attempt other techniques.
+        if len(table_names) == remaining:
+            # this is a bit tricky.
+            # run the algorithm with ONLY the declared/hard
+            # dependencies and see if it comes to resolution
+
+            higher_priority_sort = _sort_dependencies_higher_priority(
+                table_names, dependencies, priority
+            )
+
+            if higher_priority_sort:
+                sorted_tables.extend(higher_priority_sort)
+            else:
+                # I'm stuck. Try randomly
+                sorted_tables.append(sorted(table_names)[0])
+
+    return sorted_tables
+
+
+def _sort_dependencies_higher_priority(
+    table_names: T.Iterable[str],
+    dependencies: T.Mapping[str, OrderedSet],
+    priority: int,
+):
+    """After a priority-ignoring dependency sort has failed, try sorting ONLY
+    on that subset of dependencies marked with a higher priority. This function
+    is mututally recursive with `_sort_by_dependencies` so it will ratchet
+    up the priority potentially multiple times."""
+    relevant_dependencies = (
+        dep.priority
+        for depset in dependencies.values()
+        for dep in depset
+        if dep.priority > priority
+    )
+    lowest_priority_dependency = min(relevant_dependencies, default=None)
+
+    if lowest_priority_dependency is not None:
+        return _sort_by_dependencies(
+            table_names, dependencies, lowest_priority_dependency
+        )
+
+
+def _table_is_free(
+    table_name: str,
+    dependencies: StrDependencyMapping,
+    sorted_tables: T.Sequence[str],
+    priority: int,
+):
+    """Check if every child of this table is already sorted
+
+    Look at the unit test test_table_is_free_simple to see some
+    usage examples.
+    """
+    tables_this_table_depends_upon = dependencies.get(table_name, OrderedSet()).copy()
+    for dependency in sorted(tables_this_table_depends_upon):
+        if (
+            dependency.table_name_to in sorted_tables
+            or dependency.table_name_to == table_name
+            or dependency.priority < priority
+        ):
+            tables_this_table_depends_upon.remove(dependency)
+
+    return len(tables_this_table_depends_upon) == 0
+
+
+# SnowfakeryPersonAccounts: This code will be enabled when Snowfakery
+# is integrated with this code.
+# def _remove_person_contact_id(dependencies: T.Mapping[str, SObjDependency]):
+#     """Don't allow relationships between a personcontact and an account to mess
+#     up the load order.
+
+#     Code adapted from Snowfakery
+#     """
+#     if "Account" in dependencies:
+#         dep_to_person_contact = [
+#             dep
+#             for dep in dependencies["Account"]
+#             if dep.table_name_to.lower() == "personcontact"
+#         ]
+#         for dep in dep_to_person_contact:
+#             dependencies["Account"].remove(dep)
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/extract_mapping_file_generator.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/extract_mapping_file_generator.py
new file mode 100644
index 000000000..903756670
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/extract_mapping_file_generator.py
@@ -0,0 +1,41 @@
+import typing as T
+
+from cumulusci.salesforce_api.org_schema import Schema
+from cumulusci.tasks.bulkdata.generate_mapping_utils.generate_mapping_from_declarations import (
+    SimplifiedExtractDeclarationWithLookups,
+    classify_and_filter_lookups,
+)
+
+from ..extract_dataset_utils.synthesize_extract_declarations import (
+    ExtractDeclaration,
+    flatten_declarations,
+)
+
+
+def _mapping_decl_for_extract_decl(
+    decl: SimplifiedExtractDeclarationWithLookups,
+):
+    """Make a CCI extract mapping step from a SimplifiedExtractDeclarationWithLookups"""
+    lookups = {lookup: {"table": table} for lookup, table in decl.lookups.items()}
+    mapping_dict = {
+        "sf_object": decl.sf_object,
+    }
+    if decl.where:
+        mapping_dict["soql_filter"] = decl.where
+    mapping_dict["fields"] = decl.fields
+    if lookups:
+        mapping_dict["lookups"] = lookups
+
+    return (f"Extract {decl.sf_object}", mapping_dict)
+
+
+def create_extract_mapping_file_from_declarations(
+    decls: T.List[ExtractDeclaration], schema: Schema
+):
+    """Create a mapping file sufficient for driving an extract process
+    from an extract declarations file."""
+    assert decls is not None
+    simplified_decls = flatten_declarations(decls, schema)
+    simplified_decls = classify_and_filter_lookups(simplified_decls, schema)
+    mappings = [_mapping_decl_for_extract_decl(decl) for decl in simplified_decls]
+    return dict(pair for pair in mappings if pair)
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/generate_mapping_from_declarations.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/generate_mapping_from_declarations.py
new file mode 100644
index 000000000..2f214a334
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/generate_mapping_from_declarations.py
@@ -0,0 +1,111 @@
+import typing as T
+from itertools import chain
+
+from cumulusci.salesforce_api.org_schema import Schema
+from cumulusci.tasks.bulkdata.mapping_parser import MappingStep
+from cumulusci.utils.collections import OrderedSet
+from cumulusci.utils.iterators import partition
+
+from ..extract_dataset_utils.calculate_dependencies import SObjDependency
+from ..extract_dataset_utils.synthesize_extract_declarations import (
+    ExtractDeclaration,
+    SimplifiedExtractDeclaration,
+    flatten_declarations,
+)
+from .load_mapping_file_generator import generate_load_mapping_file
+
+
+class SimplifiedExtractDeclarationWithLookups(SimplifiedExtractDeclaration):
+    lookups: T.Dict[str, str]
+
+
+def create_load_mapping_file_from_extract_declarations(
+    decls: T.List[ExtractDeclaration],
+    schema: Schema,
+) -> T.Dict[str, dict]:
+    """Create a mapping file from Extract declarations"""
+    simplified_decls = flatten_declarations(decls, schema)
+    simplified_decls_w_lookups = classify_and_filter_lookups(simplified_decls, schema)
+    intertable_dependencies = _discover_dependendencies(simplified_decls_w_lookups)
+
+    def _mapping_step(decl):
+        fields = tuple(chain(decl.fields, decl.lookups.keys()))
+        return MappingStep(
+            sf_object=decl.sf_object,
+            fields=zip(fields, fields),
+            # lookups=lookups,      # lookups can be re-created later, for simplicity
+        )
+
+    mapping_steps = [_mapping_step(decl) for decl in simplified_decls_w_lookups]
+
+    mappings = generate_load_mapping_file(mapping_steps, intertable_dependencies, None)
+    return mappings
+
+
+def _discover_dependendencies(simplified_decls: T.Sequence):
+    """Look at all of the lookups in a set of declarations to determine
+    what depends on what"""
+    intertable_dependencies = OrderedSet()
+
+    for decl in simplified_decls:
+        for fieldname, tablename in decl.lookups.items():
+            intertable_dependencies.add(
+                SObjDependency(decl.sf_object, tablename, fieldname)
+            )
+    return intertable_dependencies
+
+
+def classify_and_filter_lookups(
+    decls: T.Sequence[SimplifiedExtractDeclaration], schema: Schema
+) -> T.Sequence[SimplifiedExtractDeclarationWithLookups]:
+    """Move lookups into their own field, if they reference a table we're including"""
+    referenceable_tables = [decl.sf_object for decl in decls]
+    return [_add_lookups_to_decl(decl, schema, referenceable_tables) for decl in decls]
+
+
+def _add_lookups_to_decl(
+    decl: SimplifiedExtractDeclaration,
+    schema: Schema,
+    referenceable_tables: T.Sequence[str],
+) -> SimplifiedExtractDeclarationWithLookups:
+    """Look at every declaration and check whether any of the fields it refers to
+    are actually lookups. If so, synthesize the lookups declarations."""
+    sobject_schema_info = schema[decl.sf_object]
+    fields, lookups_and_targets = _fields_and_lookups_for_decl(
+        decl, sobject_schema_info, referenceable_tables
+    )
+    new_decl_data = {
+        **dict(decl),
+        "fields": list(fields),
+        "lookups": dict(lookups_and_targets),
+    }
+    del new_decl_data["fields_"]
+    new_decl = SimplifiedExtractDeclarationWithLookups(**new_decl_data)
+    return new_decl
+
+
+def _fields_and_lookups_for_decl(decl, sobject_schema_info, referenceable_tables):
+    """Split fields versus lookups for a declaration"""
+    simple_fields, lookups = partition(
+        lambda field_name: sobject_schema_info.fields[field_name].referenceTo,
+        decl.fields,
+    )
+
+    def target_table(field_info):
+        if len(field_info.referenceTo) == 1:
+            target = field_info.referenceTo[0]
+        else:  # pragma: no cover  # TODO: Cover
+            target = "Polymorphic lookups are not supported"
+        return target
+
+    lookups = list(lookups)
+
+    lookups_and_targets = (
+        (lookup, target_table(sobject_schema_info.fields[lookup])) for lookup in lookups
+    )
+    lookups_and_targets = (
+        (lookup, table)
+        for lookup, table in lookups_and_targets
+        if table in referenceable_tables
+    )
+    return simple_fields, lookups_and_targets
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/load_mapping_file_generator.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/load_mapping_file_generator.py
new file mode 100644
index 000000000..f2a172e65
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/load_mapping_file_generator.py
@@ -0,0 +1,127 @@
+import typing as T
+
+from snowfakery.cci_mapping_files.declaration_parser import SObjectRuleDeclaration
+
+from cumulusci.tasks.bulkdata.extract_dataset_utils.calculate_dependencies import (
+    SObjDependency,
+)
+from cumulusci.tasks.bulkdata.generate_mapping_utils.dependency_map import DependencyMap
+from cumulusci.tasks.bulkdata.generate_mapping_utils.mapping_transforms import (
+    merge_matching_steps,
+    recategorize_lookups,
+    rename_record_type_fields,
+    sort_steps,
+)
+from cumulusci.tasks.bulkdata.mapping_parser import MappingStep
+from cumulusci.utils.collections import OrderedSet, OrderedSetType
+
+# from ..extract_dataset_utils.extract_yml import ExtractDeclaration
+from .mapping_generator_post_processes import add_after_statements
+
+# Note that the code in this file is also used by Snowfakery, so
+# changes need to be coordinated.
+#
+# It is also intended to someday be used by CCI to generate
+# Load Mapping Files from Load Declarations.
+
+
+# Code adapted from Snowfakery
+def generate_load_mapping_file(
+    mapping_steps: T.Sequence[MappingStep],
+    intertable_dependencies: OrderedSetType[SObjDependency],
+    load_declarations: T.List[SObjectRuleDeclaration] = None,
+) -> T.Dict[str, dict]:
+    """Generate a mapping file in optimal order with forward references etc.
+
+    Input is a set of a tables, dependencies between them and load declarations"""
+
+    load_declarations = load_declarations or []
+    declared_dependencies = collect_user_specified_dependencies(load_declarations)
+    table_names = OrderedSet(mapping.sf_object for mapping in mapping_steps)
+    depmap = DependencyMap(
+        table_names, intertable_dependencies.union(declared_dependencies)
+    )
+
+    # Merge similar steps (output step count will be equal or less)
+    mapping_steps = merge_matching_steps(mapping_steps, depmap)
+
+    # Sort steps according to dependencies
+    mapping_steps = sort_steps(mapping_steps, depmap)
+
+    # Normalize various forms of record type fields
+    mapping_steps = rename_record_type_fields(mapping_steps, depmap)
+
+    # Detect which fields are lookups and properly set them up
+    mapping_steps = recategorize_lookups(mapping_steps, depmap)
+
+    # Apply user-specified declarations
+    mapping_steps = apply_declarations(mapping_steps, load_declarations)
+
+    mappings_dict = mappings_as_dicts(mapping_steps, depmap)
+    add_after_statements(mappings_dict)
+    return mappings_dict
+
+    # note that it is entirely possible for e.g.
+    # M=10 input mapping steps to relate to N=3 tables which generate P=7 load steps.
+    #
+    # This invariiant will always hold
+    #
+    # N <= P <= M
+    #
+    # e.g. 10 Snowfakery Templates generating Accounts, Contacts, Opportunities
+    # But Accounts have 5 different update keys and the other two have none.
+
+
+def collect_user_specified_dependencies(
+    declarations: T.Iterable[SObjectRuleDeclaration],
+) -> OrderedSetType[SObjDependency]:
+    """The user can specify load order dependencies."""
+    declared_dependencies = [decl for decl in declarations if decl.load_after]
+    return OrderedSet(
+        [
+            SObjDependency(decl.sf_object, decl.load_after, "DECLARED", True)
+            for decl in declared_dependencies
+        ]
+    )
+
+
+def mappings_as_dicts(
+    load_steps: T.List[MappingStep],
+    depmap: DependencyMap,
+) -> T.Mapping[str, T.Dict]:
+    """Generate mapping.yml data structures."""
+    mappings = {}
+    for mapping_step in load_steps:
+        step_name = f"{mapping_step.action.value.title()} {mapping_step.sf_object}"
+        # Will be used by Snowfakery to generate Upsert mapping files
+        if mapping_step.update_key:
+            step_name += f" {'_'.join(mapping_step.update_key)}"
+
+        assert step_name not in mappings
+
+        mappings[step_name] = mapping_step.dict(
+            by_alias=True,
+            exclude_defaults=True,
+        )
+    return mappings
+
+
+def apply_declarations(
+    mapping_steps: T.List[MappingStep],
+    load_declarations: T.List[SObjectRuleDeclaration] = None,
+) -> T.List[MappingStep]:
+    """Apply user-specified declarations.
+    Will be used when this code is used to generate load mappings
+    from schemas and SQL instead of from ExtractDeclarations."""
+
+    load_declarations = {decl.sf_object: decl for decl in load_declarations}
+
+    def doit(mapping_step: MappingStep) -> MappingStep:
+        if sobject_declarations := load_declarations.get(mapping_step.sf_object):
+            return MappingStep(
+                **{**mapping_step.dict(), **sobject_declarations.as_mapping()}
+            )
+        else:
+            return mapping_step
+
+    return [doit(step) for step in mapping_steps]
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/mapping_generator_post_processes.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/mapping_generator_post_processes.py
new file mode 100644
index 000000000..aa2fe6213
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/mapping_generator_post_processes.py
@@ -0,0 +1,46 @@
+from typing import NamedTuple
+
+
+def add_after_statements(mappings: dict):
+    """Automatically add CCI after: statements to the lookups
+    in a mapping file"""
+    indexed_by_sobject = _index_by_sobject(mappings)
+
+    for idx, (_, mapping) in enumerate(mappings.items()):
+        for lookup in mapping.get("lookups", {}).values():
+            target_table = lookup["table"]
+            # SnowfakeryPersonAccounts: Add this back in when Snowfakery is integrated with this code.
+            # PersonContacts are not real, so skip them
+            if target_table == "PersonContact":  # pragma: no cover
+                raise NotImplementedError(
+                    "This code is not yet tested for use with Snowfakery and Person Accounts"
+                )
+                # continue
+            target_mapping_index = indexed_by_sobject[target_table]
+            if target_mapping_index.first_instance >= idx:
+                if not lookup.get("after"):
+                    lookup["after"] = target_mapping_index.last_step_name
+
+
+class MappingIndex(NamedTuple):  # info needed by the algorithm above
+    first_instance: int  # where was the first time this sobj was referenced?
+    last_step_name: str  # where was the last (so far)?
+
+
+def _index_by_sobject(mappings):
+    """Figure out the order in which sobjects are loaded and assign an index to each sobject
+
+    We can use this to detect sobjects out-of-order which require 'after' declarations."""
+    indexed_by_sobject = {}
+    for idx, (mapping_name, mapping) in enumerate(mappings.items()):
+        # make an index of the order of objects
+        sobject = mapping["sf_object"]
+        existing_index = indexed_by_sobject.get(sobject)
+
+        if existing_index:
+            new_mi = MappingIndex(existing_index.first_instance, mapping_name)
+        else:
+            new_mi = MappingIndex(idx, mapping_name)
+        indexed_by_sobject[sobject] = new_mi
+
+    return indexed_by_sobject
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/mapping_transforms.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/mapping_transforms.py
new file mode 100644
index 000000000..5393e14e4
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/mapping_transforms.py
@@ -0,0 +1,139 @@
+import itertools
+import typing as T
+
+from snowfakery.salesforce import find_record_type_column
+
+from cumulusci.tasks.bulkdata.mapping_parser import MappingStep
+from cumulusci.tasks.bulkdata.step import DataOperationType
+from cumulusci.utils.yaml.model_parser import CCIModel
+
+"""A mapping transform is a function with the signature:
+
+def func(mapping_steps: T.List[MappingStep], depmap: "DependencyMap") -> T.List[MappingStep]:
+
+It should be idempotent and a pure function (don't mutate arguments).
+
+Ideally the output should be as similar to the input as possible (i.e.
+change what you need to change and leave the rest as it was). In particular,
+the order of the input steps should be preserved unless reordering them
+is the specific task of the transform.
+"""
+
+from cumulusci.tasks.bulkdata.generate_mapping_utils.dependency_map import DependencyMap
+
+
+def sort_steps(
+    mapping_steps: T.List[MappingStep], depmap: DependencyMap
+) -> T.List[MappingStep]:
+    """Sort mapping declaration steps in dependency order"""
+    table_order = depmap.get_dependency_order()
+
+    return sorted(mapping_steps, key=lambda step: table_order.index(step.sf_object))
+
+
+class _MappingStepSignature(CCIModel):
+    index: int
+    sf_object: T.Optional[str]
+    filters: T.Optional[T.Tuple[str]] = ()
+    soql_filter: T.Optional[str] = None  # soql_filter property
+    action: DataOperationType
+    update_key: T.Union[str, T.Tuple[str, ...]] = ()  # only for upserts
+
+
+def merge_matching_steps(
+    steps: T.Iterable[MappingStep], depmap: DependencyMap
+) -> T.List[MappingStep]:
+    """Merge mapping declaration steps with similar properties"""
+    table_order = {obj.sf_object: idx for idx, obj in enumerate(steps)}
+    grouped_steps = itertools.groupby(
+        steps,
+        lambda step: _MappingStepSignature(
+            index=table_order[step.sf_object],
+            sf_object=step.sf_object,
+            filters=tuple(step.filters) or None,
+            soql_filter=step.soql_filter,
+            action=step.action,
+            update_key=step.update_key,
+        ),
+    )
+
+    new_steps = [_merge_steps(steps) for _group, steps in grouped_steps]
+    return new_steps
+
+
+def _merge_steps(steps: T.Iterable[MappingStep]) -> MappingStep:
+    """Merge a groupp of matching steps together"""
+    steps = tuple(steps)
+    new_props = steps[0].dict()
+    new_props["fields"] = {}
+    new_props["lookups"] = {}
+    for step in steps:
+        new_props["fields"].update(step.fields)
+        new_props["lookups"].update(step.lookups)
+    return MappingStep(**new_props)
+
+
+def rename_record_type_fields(
+    mapping_steps: T.List[MappingStep], depmap: DependencyMap
+) -> T.List[MappingStep]:
+    """Rename fields like recordtype, recordtypeid, recordtype_id to RecordTypeId"""
+
+    def doit(mapping_step: MappingStep):
+        table_name = mapping_step.sf_object
+        record_type_col = find_record_type_column(table_name, mapping_step.fields)
+        if record_type_col:
+            fields = mapping_step.fields.copy()
+            fields["RecordTypeId"] = fields.pop(record_type_col)
+            return MappingStep(**{**mapping_step.dict(), "fields": fields})
+        else:
+            return mapping_step
+
+    return list(map(doit, mapping_steps))
+
+
+def recategorize_lookups(
+    mapping_steps: T.List[MappingStep], depmap: DependencyMap
+) -> T.List[MappingStep]:
+    """Categorize fields which are actually lookups
+
+    Put them in mapping_step.lookup and remove them from mapping_step.fields"""
+
+    def doit(mapping_step: MappingStep):
+        table_name = mapping_step.sf_object
+        lookups = mapping_step.lookups.copy() or {}
+        lookups.update(
+            {
+                fieldname: {
+                    "table": depmap.target_table_for(table_name, fieldname),
+                    "key_field": fieldname,
+                }
+                for fieldname in mapping_step.fields
+                if depmap.target_table_for(table_name, fieldname)
+            }
+        )
+
+        if lookups:
+            fields = [
+                field for field in mapping_step.fields if field not in lookups.keys()
+            ]
+            return MappingStep(
+                **{**mapping_step.dict(), "lookups": lookups, "fields": fields}
+            )
+        else:
+            return mapping_step
+
+    return list(map(doit, mapping_steps))
+
+
+# SnowfakeryPersonAccounts: This will be turned on when Snowfakery is
+# integrated with this code.
+#
+# def change_person_contact_sf_object(
+#     mapping_steps: T.List[MappingStep], depmap: DependencyMap
+# ) -> T.List[MappingStep]:
+#     def doit(mapping_step: MappingStep):
+
+#         if mapping_step.table_name == "PersonContact":
+#             mapping_step.sf_object = "Contact"
+
+#     return list(map(doit, mapping_steps))
diff --git a/cumulusci/tasks/bulkdata/mapping_parser.py b/cumulusci/tasks/bulkdata/mapping_parser.py
index 20363f8f3..2939ffe89 100644
--- a/cumulusci/tasks/bulkdata/mapping_parser.py
+++ b/cumulusci/tasks/bulkdata/mapping_parser.py
@@ -66,6 +66,11 @@ class MappingLookup(CCIDictModel):
             + f"Tried {', '.join(guesses)}"
         )
 
+    class Config:
+        # name is an injected field (from the parent dict)
+        # so don't try to serialize it as part of the model
+        fields = {"name": {"exclude": True}}
+
 
 SHOULD_REPORT_RECORD_TYPE_DEPRECATION = True
 
@@ -106,7 +111,8 @@ class MappingStep(CCIDictModel):
     def case_normalize(cls, val):
         if isinstance(val, Enum):
             return val
-        return ENUM_VALUES.get(val.lower())
+        if val is not None:
+            return ENUM_VALUES.get(val.lower())
 
     @validator("update_key", pre=True)
     def split_update_key(cls, val):
@@ -224,7 +230,8 @@ class MappingStep(CCIDictModel):
     @validator("anchor_date")
     @classmethod
     def validate_anchor_date(cls, v):
-        return iso_to_date(v)
+        if v is not None:
+            return iso_to_date(v)
 
     @validator("record_type")
     @classmethod
@@ -238,9 +245,11 @@ class MappingStep(CCIDictModel):
     @validator("oid_as_pk")
     @classmethod
     def oid_as_pk_is_deprecated(cls, v):
-        raise ValueError(
-            "oid_as_pk is no longer supported. Include the Id field if desired."
-        )
+        if v:
+            raise ValueError(
+                "oid_as_pk is no longer supported. Include the Id field if desired."
+            )
+        return v
 
     @validator("fields_", pre=True)
     @classmethod
@@ -265,7 +274,7 @@ class MappingStep(CCIDictModel):
     @classmethod
     def fixup_lookup_names(cls, v):
         "Allow lookup objects to know the key they were attached to in the mapping file."
-        for name, lookup in v["lookups"].items():
+        for name, lookup in v.get("lookups", {}).items():
             lookup.name = name
         return v
 
@@ -533,6 +542,16 @@ class MappingStep(CCIDictModel):
     def describe_data(self, sf: Salesforce):
         return describe_data(self.sf_object, sf)
 
+    def dict(self, by_alias=True, exclude_defaults=True, **kwargs):
+        out = super().dict(
+            by_alias=by_alias, exclude_defaults=exclude_defaults, **kwargs
+        )
+        if fields := out.get("fields"):
+            keys = list(fields.keys())
+            if keys == list(fields.values()):
+                out["fields"] = keys
+        return out
+
 
 class MappingSteps(CCIDictModel):
     "Mapping of named steps"
diff --git a/cumulusci/tasks/bulkdata/step.py b/cumulusci/tasks/bulkdata/step.py
index 7a8c71cbd..0356b6cea 100644
--- a/cumulusci/tasks/bulkdata/step.py
+++ b/cumulusci/tasks/bulkdata/step.py
@@ -20,6 +20,7 @@ from cumulusci.utils.xml import lxml_parse_string
 DEFAULT_BULK_BATCH_SIZE = 10_000
 DEFAULT_REST_BATCH_SIZE = 200
 MAX_REST_BATCH_SIZE = 200
+csv.field_size_limit(2**27)  # 128 MB
 
 
 class DataOperationType(Enum):
diff --git a/cumulusci/utils/collections.py b/cumulusci/utils/collections.py
new file mode 100644
index 000000000..f060fd6f8
--- /dev/null
+++ b/cumulusci/utils/collections.py
@@ -0,0 +1,32 @@
+import typing as T
+
+
+class OrderedSet(dict):
+    """Extremely minimal ordered set
+
+    Use with care, because very few methods are implemented."""
+
+    def __init__(self, other: T.Optional[T.Iterable] = None):
+        if other is not None:
+            self.update(other)
+
+    def add(self, item):
+        self[item] = item
+
+    def remove(self, item):
+        del self[item]
+
+    def copy(self):
+        return OrderedSet(self)
+
+    def update(self, other: T.Iterable):
+        super().update((x, x) for x in other)
+
+    def union(self, other: T.Iterable):
+        new = OrderedSet(self)
+        new.update(other)
+        return new
+
+
+# Workaround for Python 3.8
+OrderedSetType = T.Iterable


Test patch:
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_generate_extract_mapping_from_declarations.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_generate_extract_mapping_from_declarations.py
new file mode 100644
index 000000000..46cb7aea0
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_generate_extract_mapping_from_declarations.py
@@ -0,0 +1,74 @@
+from cumulusci.tasks.bulkdata.extract_dataset_utils.extract_yml import (
+    ExtractDeclaration,
+)
+from cumulusci.tasks.bulkdata.extract_dataset_utils.tests.test_synthesize_extract_declarations import (
+    _fake_get_org_schema,
+    describe_for,
+)
+from cumulusci.tasks.bulkdata.generate_mapping_utils.extract_mapping_file_generator import (
+    create_extract_mapping_file_from_declarations,
+)
+
+
+class TestGenerateLoadMappingFromDeclarations:
+    def test_simple_generate_mapping_from_declarations(self, org_config):
+        declarations = [
+            ExtractDeclaration(sf_object="Account", fields=["Name", "Description"])
+        ]
+        object_counts = {"Account": 10, "Contact": 0, "Case": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Case"),
+        )
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            filters=[],
+            include_counts=True,
+        ) as schema:
+            mf = create_extract_mapping_file_from_declarations(declarations, schema)
+            assert mf == {
+                "Extract Account": {
+                    "sf_object": "Account",
+                    "fields": ["Name", "Description"],
+                    "soql_filter": "Name != 'Sample Account for " "Entitlements'",
+                }
+            }
+
+    def test_generate_mapping_from_declarations__lookups(self, org_config):
+        declarations = [
+            ExtractDeclaration(sf_object="Account", fields=["Name", "Description"]),
+            ExtractDeclaration(sf_object="Contact", fields=["LastName", "AccountId"]),
+        ]
+        object_counts = {"Account": 10, "Contact": 5, "Case": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Case"),
+        )
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            filters=[],
+            include_counts=True,
+        ) as schema:
+            mf = create_extract_mapping_file_from_declarations(declarations, schema)
+            print(mf)
+            assert mf == {
+                "Extract Account": {
+                    "sf_object": "Account",
+                    "soql_filter": "Name != 'Sample Account for Entitlements'",
+                    "fields": ["Name", "Description"],
+                },
+                "Extract Contact": {
+                    "sf_object": "Contact",
+                    "fields": ["LastName"],
+                    "lookups": {"AccountId": {"table": "Account"}},
+                },
+            }
+
+
+# TODO: Figure out where clauses
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_generate_load_mapping_from_declarations.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_generate_load_mapping_from_declarations.py
new file mode 100644
index 000000000..5996d819f
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_generate_load_mapping_from_declarations.py
@@ -0,0 +1,215 @@
+from snowfakery.cci_mapping_files.declaration_parser import SObjectRuleDeclaration
+
+from cumulusci.tasks.bulkdata.extract_dataset_utils.extract_yml import (
+    ExtractDeclaration,
+)
+from cumulusci.tasks.bulkdata.extract_dataset_utils.tests.test_synthesize_extract_declarations import (
+    _fake_get_org_schema,
+    describe_for,
+)
+from cumulusci.tasks.bulkdata.generate_mapping_utils.generate_mapping_from_declarations import (
+    create_load_mapping_file_from_extract_declarations,
+    generate_load_mapping_file,
+)
+from cumulusci.tasks.bulkdata.mapping_parser import MappingStep
+from cumulusci.tasks.bulkdata.step import DataApi, DataOperationType
+
+
+class TestGenerateLoadMappingFromDeclarations:
+    def test_simple_generate_mapping_from_declarations(self, org_config):
+        declarations = [
+            ExtractDeclaration(sf_object="Account", fields=["Name", "Description"])
+        ]
+        object_counts = {"Account": 10, "Contact": 0, "Case": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Case"),
+        )
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            filters=[],
+            include_counts=True,
+        ) as schema:
+            mf = create_load_mapping_file_from_extract_declarations(
+                declarations, schema
+            )
+            assert mf == {
+                "Insert Account": {
+                    "sf_object": "Account",
+                    "table": "Account",
+                    "fields": ["Name", "Description"],
+                }
+            }
+
+    def test_generate_load_mapping_from_declarations__lookups(self, org_config):
+        declarations = [
+            ExtractDeclaration(sf_object="Account", fields=["Name", "Description"]),
+            ExtractDeclaration(
+                sf_object="Contact", fields=["FirstName", "LastName", "AccountId"]
+            ),
+        ]
+        object_counts = {"Account": 10, "Contact": 5, "Case": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Case"),
+        )
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            filters=[],
+            include_counts=True,
+        ) as schema:
+            mf = create_load_mapping_file_from_extract_declarations(
+                declarations, schema
+            )
+            assert mf == {
+                "Insert Account": {
+                    "sf_object": "Account",
+                    "table": "Account",
+                    "fields": ["Name", "Description"],
+                },
+                "Insert Contact": {
+                    "sf_object": "Contact",
+                    "table": "Contact",
+                    "fields": ["FirstName", "LastName"],
+                    "lookups": {
+                        "AccountId": {"table": "Account", "key_field": "AccountId"}
+                    },
+                },
+            }
+
+    def test_generate_load_mapping_from_declarations__circular_lookups(
+        self, org_config
+    ):
+        declarations = [
+            ExtractDeclaration(
+                sf_object="Account",
+                fields=["Name", "Description", "Primary_Contact__c"],
+            ),
+            ExtractDeclaration(
+                sf_object="Contact", fields=["FirstName", "LastName", "AccountId"]
+            ),
+        ]
+        object_counts = {"Account": 10, "Contact": 5, "Case": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Case"),
+        )
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            filters=[],
+            include_counts=True,
+        ) as schema:
+            mf = create_load_mapping_file_from_extract_declarations(
+                declarations, schema
+            )
+            assert mf == {
+                "Insert Account": {
+                    "fields": ["Name", "Description"],
+                    "lookups": {
+                        "Primary_Contact__c": {
+                            "after": "Insert Contact",
+                            "key_field": "Primary_Contact__c",
+                            "table": "Contact",
+                        }
+                    },
+                    "sf_object": "Account",
+                    "table": "Account",
+                },
+                "Insert Contact": {
+                    "sf_object": "Contact",
+                    "table": "Contact",
+                    "fields": ["FirstName", "LastName"],
+                    "lookups": {
+                        "AccountId": {"table": "Account", "key_field": "AccountId"}
+                    },
+                },
+            }, mf
+
+    def test_generate_load_mapping__with_load_declarations(self, org_config):
+
+        mapping_steps = [
+            MappingStep(sf_object="Account"),
+            MappingStep(sf_object="Contact"),
+        ]
+        load_declarations = [
+            SObjectRuleDeclaration(sf_object="Account", api="rest"),
+            SObjectRuleDeclaration(sf_object="Contact", api="bulk"),
+        ]
+        mf = generate_load_mapping_file(
+            mapping_steps,
+            intertable_dependencies=set(),
+            load_declarations=load_declarations,
+        )
+        assert mf == {
+            "Insert Account": {
+                "sf_object": "Account",
+                "api": DataApi.REST,
+                "table": "Account",
+            },
+            "Insert Contact": {
+                "sf_object": "Contact",
+                "api": DataApi.BULK,
+                "table": "Contact",
+            },
+        }, mf
+
+    def test_generate_load_mapping__with_userts(self, org_config):
+
+        mapping_steps = [
+            MappingStep(sf_object="Account"),
+            MappingStep(
+                sf_object="Account",
+                action="UPSERT",
+                update_key=["Name"],
+                fields=["Name"],
+            ),
+            MappingStep(
+                sf_object="Account",
+                action="ETL_UPSERT",
+                update_key=["AccountNumber", "Name"],
+                fields=["AccountNumber", "Name"],
+            ),
+            MappingStep(sf_object="Contact"),
+        ]
+
+        mf = generate_load_mapping_file(
+            mapping_steps,
+            intertable_dependencies=set(),
+            load_declarations=None,
+        )
+        assert mf == {
+            "Insert Account": {
+                "sf_object": "Account",
+                "table": "Account",
+            },
+            "Upsert Account Name": {
+                "sf_object": "Account",
+                "table": "Account",
+                "action": DataOperationType.UPSERT,
+                "update_key": ("Name",),
+                "fields": ["Name"],
+            },
+            "Etl_Upsert Account AccountNumber_Name": {
+                "sf_object": "Account",
+                "table": "Account",
+                "action": DataOperationType.ETL_UPSERT,
+                "update_key": ("AccountNumber", "Name"),
+                "fields": ["AccountNumber", "Name"],
+            },
+            "Insert Contact": {
+                "sf_object": "Contact",
+                "table": "Contact",
+            },
+        }, mf
+
+
+# TODO: Figure out where clauses
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_mapping_generator_post_processes.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_mapping_generator_post_processes.py
new file mode 100644
index 000000000..e01b86ce6
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_mapping_generator_post_processes.py
@@ -0,0 +1,29 @@
+from cumulusci.tasks.bulkdata.generate_mapping_utils.mapping_generator_post_processes import (
+    add_after_statements,
+)
+from cumulusci.tasks.bulkdata.mapping_parser import MappingLookup, MappingStep
+
+
+class TestMappingGeneratorPostProcesses:
+    def test_add_after_statements(self):
+        "Test that the add_after_statements function will add an `after` statement to the correct mapping step"
+        mappings = {
+            "Insert Blah": MappingStep(sf_object="Blah"),
+            "Insert Accounts": MappingStep(
+                sf_object="Account",
+                lookups={"PrimaryContact__c": MappingLookup(table="Contact")},
+            ),
+            "Update Account": MappingStep(
+                sf_object="Account",
+                action="Update",
+            ),
+            "Insert Contact": MappingStep(
+                sf_object="Contact",
+                lookups={"AccountId": MappingLookup(table="Account")},
+            ),
+        }
+        add_after_statements(mappings)
+        assert (
+            mappings["Insert Accounts"].lookups["PrimaryContact__c"].after
+            == "Insert Contact"
+        )
diff --git a/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_mapping_transforms.py b/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_mapping_transforms.py
new file mode 100644
index 000000000..36bd3fdcc
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/generate_mapping_utils/tests/test_mapping_transforms.py
@@ -0,0 +1,188 @@
+import random
+
+from cumulusci.tasks.bulkdata.extract_dataset_utils.calculate_dependencies import (
+    SObjDependency,
+)
+from cumulusci.tasks.bulkdata.generate_mapping_utils.dependency_map import DependencyMap
+from cumulusci.tasks.bulkdata.generate_mapping_utils.mapping_transforms import (
+    merge_matching_steps,
+    recategorize_lookups,
+    rename_record_type_fields,
+    sort_steps,
+)
+from cumulusci.tasks.bulkdata.mapping_parser import MappingStep
+
+
+class MappingTransformTesterBase:
+    @classmethod
+    def setup_class(_):
+        mttb = MappingTransformTesterBase
+        mttb.std_intertable_dependencies = [
+            SObjDependency("Contact", "Account", "AccountId"),
+            SObjDependency("Account", "Account", "ParentId"),
+            SObjDependency("Opportunity", "Contact", "ContactId"),
+            SObjDependency("Opportunity", "Account", "AccountId"),
+            SObjDependency("Account", "Custom__c", "Custom__c"),
+            SObjDependency("Custom__c", "Opportunity", "Opportunity__c"),
+            SObjDependency("Custom_2__c", "Custom__c", "Custom__c"),
+            SObjDependency("Custom_2__c", "Custom_3__c", "Custom_3__c"),
+            SObjDependency("Custom_3__c", "Custom_2__c", "Custom_2__c"),
+        ]
+        mttb.std_table_names = [
+            dep.table_name_from for dep in mttb.std_intertable_dependencies
+        ] + ["Unrelated__c"]
+
+        mttb.std_dep_map = DependencyMap(
+            mttb.std_table_names, mttb.std_intertable_dependencies
+        )
+
+
+class TestSortSteps(MappingTransformTesterBase):
+    def test_sort_steps(self):
+        steps = [
+            MappingStep(sf_object="Opportunity"),
+            MappingStep(sf_object="Custom__c"),
+            MappingStep(sf_object="Custom_2__c"),
+            MappingStep(sf_object="Custom_3__c"),
+            MappingStep(sf_object="Contact"),
+            MappingStep(sf_object="Account"),
+        ]
+
+        for i in range(0, 10):
+            random.Random(i).shuffle(steps)
+            out_steps = sort_steps(steps, self.std_dep_map)
+            enumerated = enumerate(step.sf_object for step in out_steps)
+            positions = {name: pos for (pos, name) in enumerated}
+            assert positions["Account"] < positions["Contact"], positions
+            assert positions["Account"] < positions["Opportunity"], positions
+            assert positions["Contact"] < positions["Opportunity"], positions
+            assert positions["Opportunity"] < positions["Custom__c"], positions
+            assert positions["Opportunity"] < positions["Custom__c"], positions
+            assert positions["Custom__c"] < positions["Custom_2__c"], positions
+            assert positions["Custom__c"] < positions["Custom_3__c"], positions
+
+    def test_sort_steps__with_priorities(self):
+        steps = [
+            MappingStep(sf_object="Opportunity"),
+            MappingStep(sf_object="Custom__c"),
+            MappingStep(sf_object="Custom_2__c"),
+            MappingStep(sf_object="Custom_3__c"),
+            MappingStep(sf_object="Contact"),
+            MappingStep(sf_object="Account"),
+        ]
+
+        for i in range(0, 10):
+            random.Random(i).shuffle(steps)
+            # add a high priority requirement that forces Custom_3__c to go before Account
+            deps = DependencyMap(
+                self.std_table_names,
+                self.std_intertable_dependencies
+                + [SObjDependency("Account", "Custom_3__c", "Custom_3__c", True)],
+            )
+            out_steps = sort_steps(steps, deps)
+            enumerated = enumerate(step.sf_object for step in out_steps)
+            positions = {name: pos for (pos, name) in enumerated}
+            assert positions["Custom_3__c"] < positions["Account"], positions
+            assert positions["Opportunity"] < positions["Custom__c"], positions
+            assert positions["Opportunity"] < positions["Custom__c"], positions
+            assert positions["Custom__c"] < positions["Custom_2__c"], positions
+            assert positions["Custom__c"] < positions["Custom_3__c"], positions
+
+    def test_sort_steps__with_priorities__2(self):
+        steps = [
+            MappingStep(sf_object="Opportunity"),
+            MappingStep(sf_object="Custom__c"),
+            MappingStep(sf_object="Custom_2__c"),
+            MappingStep(sf_object="Custom_3__c"),
+            MappingStep(sf_object="Contact"),
+            MappingStep(sf_object="Account"),
+        ]
+
+        for i in range(0, 10):
+            random.Random(i).shuffle(steps)
+            # add a high priority requirement that forces Custom_2__c to go before Custom_3__c
+            deps = DependencyMap(
+                self.std_table_names,
+                self.std_intertable_dependencies
+                + [SObjDependency("Custom_3__c", "Custom_2__c", "Custom_2__c", True)],
+            )
+            out_steps = sort_steps(steps, deps)
+            enumerated = enumerate(step.sf_object for step in out_steps)
+            positions = {name: pos for (pos, name) in enumerated}
+            assert positions["Custom_2__c"] < positions["Custom_3__c"], positions
+
+
+class TestMergeMatchingSteps(MappingTransformTesterBase):
+    def test_merge_matching_steps(self):
+        steps = [
+            MappingStep(sf_object="Account", fields={"recordtypeid": "recordtypeid"}),
+            MappingStep(sf_object="Account", fields={"Name": "Name"}),
+            MappingStep(sf_object="Contact", fields={"recordtype": "recordtype"}),
+            MappingStep(sf_object="Contact", fields={"description": "description"}),
+        ]
+        out_steps = merge_matching_steps(steps, self.std_dep_map)
+        by_name = {step.sf_object: step for step in out_steps}
+        assert tuple(step.sf_object for step in out_steps) == ("Account", "Contact")
+        assert tuple(by_name["Account"].fields) == ("recordtypeid", "Name")
+        assert tuple(by_name["Contact"].fields) == ("recordtype", "description")
+
+
+class TestRenameRecordTypeFields(MappingTransformTesterBase):
+    def test_rename_record_type_fields(self):
+        steps = [
+            MappingStep(
+                sf_object="Account", fields={"record_type_id": "record_type_id"}
+            ),
+            MappingStep(sf_object="Custom__c", fields={"recordtypeid": "recordtypeid"}),
+            MappingStep(sf_object="Custom_2__c", fields={"recordtype": "recordtype"}),
+            MappingStep(sf_object="Custom_3__c"),
+        ]
+        out_steps = rename_record_type_fields(steps, self.std_dep_map)
+
+        by_name = {step.sf_object: step for step in out_steps}
+        assert tuple(by_name.keys()) == (
+            "Account",
+            "Custom__c",
+            "Custom_2__c",
+            "Custom_3__c",
+        )
+
+        assert "RecordTypeId" in by_name["Account"].fields
+        assert "RecordTypeId" in by_name["Custom__c"].fields
+        assert "RecordTypeId" in by_name["Custom_2__c"].fields
+        assert "RecordTypeId" not in by_name["Custom_3__c"].fields
+
+
+class TestRecategorizeLookups(MappingTransformTesterBase):
+    def test_recategorize_lookups(self):
+        steps = [
+            MappingStep(
+                sf_object="Account",
+                fields={
+                    "ParentId": "ParentId",
+                    "Name": "Name",
+                    "Custom__c": "Custom__c",
+                },
+            ),
+            MappingStep(
+                sf_object="Contact",
+                fields={"AccountId": "AccountId", "Firstname": "Firstname"},
+            ),
+            MappingStep(
+                sf_object="Custom_2__c",
+                fields={"Custom__c": "Custom__c", "Name": "Name"},
+            ),
+            MappingStep(
+                sf_object="Custom_3__c",
+                fields={"Name": "Name"},
+            ),
+        ]
+        out_steps = recategorize_lookups(steps, self.std_dep_map)
+
+        by_name = {step.sf_object: step for step in out_steps}
+        assert tuple(by_name["Account"].fields) == ("Name",)
+        assert tuple(by_name["Account"].lookups) == ("ParentId", "Custom__c")
+        assert tuple(by_name["Contact"].fields) == ("Firstname",)
+        assert tuple(by_name["Contact"].lookups) == ("AccountId",)
+        assert tuple(by_name["Custom_2__c"].fields) == ("Name",)
+        assert tuple(by_name["Custom_2__c"].lookups) == ("Custom__c",)
diff --git a/cumulusci/tasks/bulkdata/tests/test_mapping_parser.py b/cumulusci/tasks/bulkdata/tests/test_mapping_parser.py
index 1b5c8b9da..afa397d09 100644
--- a/cumulusci/tasks/bulkdata/tests/test_mapping_parser.py
+++ b/cumulusci/tasks/bulkdata/tests/test_mapping_parser.py
@@ -1174,6 +1174,34 @@ class TestMappingLookup:
         assert mapping["Insert Accounts"].action.value == "insert"
         assert mapping["Insert Accounts"].batch_size == 50
 
+    def test_oid_as_pk__raises(self):
+        with pytest.raises(ValueError):
+            parse_from_yaml(
+                StringIO(
+                    (
+                        """Insert Accounts:
+                            sf_object: account
+                            oid_as_pk: True
+                            fields:
+                                - name"""
+                    )
+                )
+            )
+
+    def test_oid_as_pk__false(self):
+        result = parse_from_yaml(
+            StringIO(
+                (
+                    """Insert Accounts:
+                            sf_object: account
+                            oid_as_pk: False
+                            fields:
+                                - name"""
+                )
+            )
+        )
+        assert result["Insert Accounts"].oid_as_pk is False
+
 
 class TestUpsertKeyValidations:
     def test_upsert_key_wrong_type(self):