diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index dfabccd6b0..d9fec078d1 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -3210,7 +3210,30 @@ definitions:
         additionalProperties: true
   KeysReplace:
     title: Keys Replace
-    description: A transformation that replaces symbols in keys.
+    description: >-
+      A transformation that replaces symbols in keys. By default, every occurrence of `old` in each key is
+      replaced with `new` (literal match), the old key is dropped (rename), and the replacement is applied
+      recursively to nested objects of the whole record. `regex`, `keep_original`, `only_if_missing` and
+      `field_path` adjust this behavior. Keys are evaluated against a snapshot of the object, so keys written
+      by the transformation are not processed again. It can also be used in `DynamicSchemaLoader.schema_transformations`,
+      where it is applied to the map of schema properties.
+    examples:
+      - type: KeysReplace
+        old: " "
+        new: "_"
+      - type: KeysReplace
+        old: "hs_v2_date_entered_"
+        new: "hs_date_entered_"
+        keep_original: true
+        only_if_missing: true
+        field_path: ["properties"]
+      - type: KeysReplace
+        old: "hs_v2_date_entered_(.*?)(?:_date)?$"
+        new: "hs_lifecyclestage_\\1_date"
+        regex: true
+        keep_original: true
+        only_if_missing: true
+        field_path: ["properties"]
     type: object
     required:
       - type
@@ -3246,6 +3269,39 @@ definitions:
           - config
           - record
           - stream_slice
+      regex:
+        type: boolean
+        title: Regex
+        description: >-
+          If true, `old` is treated as a regular expression pattern and `new` as a `re.sub` replacement string,
+          which supports backreferences such as `\1` or `\g<1>`. Invalid patterns fail when the component is created.
+        default: false
+      keep_original:
+        type: boolean
+        title: Keep Original
+        description: If true, the value is copied to the new key and the original key is kept, instead of renaming it.
+        default: false
+      only_if_missing:
+        type: boolean
+        title: Only If Missing
+        description: >-
+          If true, the new key is only written when it is absent from the object or its value is null. When the key
+          is not written and `keep_original` is false, the original key is kept unchanged.
+        default: false
+      field_path:
+        type: array
+        title: Field Path
+        description: >-
+          Path to the object whose keys are replaced. Defaults to the whole record. A `*` segment applies the
+          replacement to every matching object. If the path does not exist or does not point to an object,
+          the transformation does nothing.
+        items:
+          type: string
+        interpolation_context:
+          - config
+        examples:
+          - ["properties"]
+          - ["data", "*", "attributes"]
       $parameters:
         type: object
         additionalProperties: true
diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
index 5581be3757..1faaaa21b9 100644
--- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
+++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -842,6 +842,27 @@ class KeysReplace(BaseModel):
         ],
         title="New value",
     )
+    regex: Optional[bool] = Field(
+        False,
+        description="If true, `old` is treated as a regular expression pattern and `new` as a `re.sub` replacement string, which supports backreferences such as `\\1` or `\\g<1>`. Invalid patterns fail when the component is created.",
+        title="Regex",
+    )
+    keep_original: Optional[bool] = Field(
+        False,
+        description="If true, the value is copied to the new key and the original key is kept, instead of renaming it.",
+        title="Keep Original",
+    )
+    only_if_missing: Optional[bool] = Field(
+        False,
+        description="If true, the new key is only written when it is absent from the object or its value is null. When the key is not written and `keep_original` is false, the original key is kept unchanged.",
+        title="Only If Missing",
+    )
+    field_path: Optional[List[str]] = Field(
+        None,
+        description="Path to the object whose keys are replaced. Defaults to the whole record. A `*` segment applies the replacement to every matching object. If the path does not exist or does not point to an object, the transformation does nothing.",
+        examples=[["properties"], ["data", "*", "attributes"]],
+        title="Field Path",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
index 97a1f4ce2d..9f19ddee64 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -1103,7 +1103,13 @@ def create_keys_replace_transformation(
         self, model: KeysReplaceModel, config: Config, **kwargs: Any
     ) -> KeysReplaceTransformation:
         return KeysReplaceTransformation(
-            old=model.old, new=model.new, parameters=model.parameters or {}
+            old=model.old,
+            new=model.new,
+            regex=bool(model.regex),
+            keep_original=bool(model.keep_original),
+            only_if_missing=bool(model.only_if_missing),
+            field_path=list(model.field_path) if model.field_path else None,
+            parameters=model.parameters or {},
         )
 
     def create_flatten_fields(
diff --git a/airbyte_cdk/sources/declarative/transformations/keys_replace_transformation.py b/airbyte_cdk/sources/declarative/transformations/keys_replace_transformation.py
index 45766e70e2..f5aa9f4cc8 100644
--- a/airbyte_cdk/sources/declarative/transformations/keys_replace_transformation.py
+++ b/airbyte_cdk/sources/declarative/transformations/keys_replace_transformation.py
@@ -2,14 +2,22 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 #
 
+import copy
+import re
 from dataclasses import InitVar, dataclass
-from typing import Any, Dict, Mapping, Optional
+from typing import Any, Dict, List, Mapping, Optional, Pattern, Union
+
+import dpath
 
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.transformations import RecordTransformation
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
 
 
+def _is_interpolated(value: str) -> bool:
+    return "{{" in value or "{%" in value
+
+
 @dataclass
 class KeysReplaceTransformation(RecordTransformation):
     """
@@ -22,15 +30,64 @@ class KeysReplaceTransformation(RecordTransformation):
     Result:
     from: {"created time": ..., "customer id": ..., "user id": ...}
     to: {"created_time": ..., "customer_id": ..., "user_id": ...}
+
+    By default, `old` is matched literally, matching keys are renamed and the replacement is applied
+    recursively to nested objects of the whole record. Optional flags change this behavior:
+    - regex: `old` is a regular expression and `new` a `re.sub` replacement (backreferences supported).
+    - keep_original: the value is copied to the new key and the original key is kept.
+    - only_if_missing: the new key is only written when it is absent or null.
+    - field_path: the replacement is scoped to the object at this path (`*` matches every element).
+    Keys are evaluated against a snapshot of each object, so keys written by the transformation are not
+    processed again.
     """
 
     old: str
     new: str
     parameters: InitVar[Mapping[str, Any]]
+    regex: bool = False
+    keep_original: bool = False
+    only_if_missing: bool = False
+    field_path: Optional[List[Union[InterpolatedString, str]]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._old = InterpolatedString.create(self.old, parameters=parameters)
         self._new = InterpolatedString.create(self.new, parameters=parameters)
+        self._field_path = [
+            InterpolatedString.create(path, parameters=parameters)
+            for path in (self.field_path or [])
+        ]
+        self._static_pattern: Optional[Pattern[str]] = None
+        if self.regex and not _is_interpolated(self.old):
+            self._static_pattern = self._compile_pattern(self.old)
+            if not _is_interpolated(self.new):
+                self._validate_replacement(self._static_pattern, self.new)
+
+    @staticmethod
+    def _compile_pattern(pattern: str) -> Pattern[str]:
+        try:
+            return re.compile(pattern)
+        except re.error as exception:
+            raise ValueError(
+                f"KeysReplace `old` is not a valid regular expression: {pattern!r} ({exception})."
+            ) from exception
+
+    @staticmethod
+    def _validate_replacement(pattern: Pattern[str], replacement: str) -> None:
+        try:
+            pattern.sub(replacement, "")
+        except (re.error, IndexError) as exception:
+            raise ValueError(
+                f"KeysReplace `new` is not a valid replacement for pattern {pattern.pattern!r}: {replacement!r} ({exception})."
+            ) from exception
+
+    def _eval_regex_part(
+        self, raw: str, interpolated: InterpolatedString, config: Config, **kwargs: Any
+    ) -> str:
+        # Plain strings are used as-is: Jinja evaluation runs `ast.literal_eval` on the output,
+        # which would turn a pattern like `(1)` into `1`.
+        if not _is_interpolated(raw):
+            return raw
+        return str(interpolated.eval(config, valid_types=(str,), **kwargs))
 
     def transform(
         self,
@@ -43,19 +100,67 @@ def transform(
             config = {}
 
         kwargs = {"record": record, "stream_state": stream_state, "stream_slice": stream_slice}
-        old_key = str(self._old.eval(config, **kwargs))
-        new_key = str(self._new.eval(config, **kwargs))
+        if self.regex:
+            pattern = self._static_pattern or self._compile_pattern(
+                self._eval_regex_part(self.old, self._old, config, **kwargs)
+            )
+            replacement = self._eval_regex_part(self.new, self._new, config, **kwargs)
+            if self._static_pattern is None or _is_interpolated(self.new):
+                self._validate_replacement(pattern, replacement)
+
+            def replace_key(key: str) -> str:
+                return pattern.sub(replacement, key)
+        else:
+            old_key = str(self._old.eval(config, **kwargs))
+            new_key = str(self._new.eval(config, **kwargs))
+
+            def replace_key(key: str) -> str:
+                return key.replace(old_key, new_key)
 
         def _transform(data: Dict[str, Any]) -> Dict[str, Any]:
+            transformed = {
+                key: _transform(value) if isinstance(value, dict) else value
+                for key, value in data.items()
+            }
+
+            if self.keep_original:
+                result = dict(transformed)
+                for key, value in transformed.items():
+                    updated_key = replace_key(key)
+                    if updated_key == key:
+                        continue
+                    if self.only_if_missing and result.get(updated_key) is not None:
+                        continue
+                    result[updated_key] = (
+                        copy.deepcopy(value) if isinstance(value, (dict, list)) else value
+                    )
+                return result
+
             result = {}
-            for key, value in data.items():
-                updated_key = key.replace(old_key, new_key)
-                if isinstance(value, dict):
-                    result[updated_key] = _transform(value)
-                else:
-                    result[updated_key] = value
+            for key, value in transformed.items():
+                updated_key = replace_key(key)
+                if self.only_if_missing:
+                    if updated_key != key and (
+                        result.get(updated_key) is not None
+                        or transformed.get(updated_key) is not None
+                    ):
+                        updated_key = key
+                    if value is None and result.get(updated_key) is not None:
+                        continue
+                result[updated_key] = value
             return result
 
-        transformed_record = _transform(record)
-        record.clear()
-        record.update(transformed_record)
+        for target in self._get_targets(record, config):
+            transformed_target = _transform(target)
+            target.clear()
+            target.update(transformed_target)
+
+    def _get_targets(self, record: Dict[str, Any], config: Config) -> List[Dict[str, Any]]:
+        if not self._field_path:
+            return [record]
+        path = [str(path.eval(config)) for path in self._field_path]
+        if "*" in path:
+            matches = dpath.values(record, path)
+        else:
+            matches = [dpath.get(record, path, default=None)]
+        return [match for match in matches if isinstance(match, dict)]
diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py
index 56e1ad76f0..6183b7a7f7 100644
--- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py
+++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py
@@ -144,6 +144,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     JsonlDecoder as JsonlDecoderModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    KeysReplace as KeysReplaceModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     OffsetIncrement as OffsetIncrementModel,
 )
@@ -10096,3 +10099,77 @@ def test_create_file_uploader_with_a_combined_download_target_extractor():
     )
 
     assert isinstance(file_uploader.download_target_extractor, CombinedExtractor)
+
+
+def test_create_keys_replace_transformation_defaults():
+    transformation = factory.create_component(
+        model_type=KeysReplaceModel,
+        component_definition={"type": "KeysReplace", "old": " ", "new": "_"},
+        config=input_config,
+    )
+
+    assert isinstance(transformation, KeysReplaceTransformation)
+    assert transformation.regex is False
+    assert transformation.keep_original is False
+    assert transformation.only_if_missing is False
+    assert transformation.field_path is None
+
+
+def test_create_keys_replace_transformation_with_options():
+    content = r"""
+    type: KeysReplace
+    old: "hs_v2_date_entered_(.*?)(?:_date)?$"
+    new: "hs_lifecyclestage_\\1_date"
+    regex: true
+    keep_original: true
+    only_if_missing: true
+    field_path: ["properties"]
+    """
+    definition = YamlDeclarativeSource._parse(content)
+
+    transformation = factory.create_component(
+        model_type=KeysReplaceModel, component_definition=definition, config=input_config
+    )
+
+    assert isinstance(transformation, KeysReplaceTransformation)
+    assert transformation.new == r"hs_lifecyclestage_\1_date"
+    assert transformation.regex is True
+    assert transformation.keep_original is True
+    assert transformation.only_if_missing is True
+    assert transformation.field_path == ["properties"]
+
+    record = {
+        "id": 1,
+        "properties": {
+            "hs_v2_date_entered_customer": "2024-01-01",
+            "hs_v2_date_entered_lead_date": "2024-02-01",
+            "hs_v2_date_entered_opportunity": "2024-03-01",
+            "hs_lifecyclestage_opportunity_date": "2023-12-31",
+        },
+    }
+    transformation.transform(record, config=input_config)
+    assert record == {
+        "id": 1,
+        "properties": {
+            "hs_v2_date_entered_customer": "2024-01-01",
+            "hs_v2_date_entered_lead_date": "2024-02-01",
+            "hs_v2_date_entered_opportunity": "2024-03-01",
+            "hs_lifecyclestage_customer_date": "2024-01-01",
+            "hs_lifecyclestage_lead_date": "2024-02-01",
+            "hs_lifecyclestage_opportunity_date": "2023-12-31",
+        },
+    }
+
+
+def test_create_keys_replace_transformation_with_invalid_regex_raises():
+    with pytest.raises(ValueError, match="KeysReplace `old` is not a valid regular expression"):
+        factory.create_component(
+            model_type=KeysReplaceModel,
+            component_definition={
+                "type": "KeysReplace",
+                "old": "(unclosed",
+                "new": "x",
+                "regex": True,
+            },
+            config=input_config,
+        )
diff --git a/unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py b/unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py
index ec20333079..56e289b0ab 100644
--- a/unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py
+++ b/unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py
@@ -313,6 +313,53 @@ def test_dynamic_schema_loader_manifest_flow():
     assert actual_catalog.streams[0].json_schema == expected_schema
 
 
+def test_dynamic_schema_loader_manifest_flow_with_keys_replace():
+    manifest = deepcopy(_MANIFEST)
+    schema_loader = manifest["definitions"]["party_members_stream"]["schema_loader"]
+    schema_loader["retriever"]["requester"]["path"] = "/keys_replace/schema"
+    schema_loader["schema_transformations"] = [
+        {
+            "type": "KeysReplace",
+            "old": "hs_v2_date_entered_(.*?)(?:_date)?$",
+            "new": "hs_lifecyclestage_\\1_date",
+            "regex": True,
+            "keep_original": True,
+            "only_if_missing": True,
+        }
+    ]
+    expected_properties = {
+        "hs_v2_date_entered_customer": {"type": ["null", "string"]},
+        "hs_v2_date_entered_lead_date": {"type": ["null", "string"]},
+        "hs_lifecyclestage_lead_date": {"type": ["null", "integer"]},
+        "hs_lifecyclestage_customer_date": {"type": ["null", "string"]},
+    }
+
+    source = ConcurrentDeclarativeSource(
+        source_config=manifest, config=_CONFIG, catalog=None, state=None
+    )
+
+    with HttpMocker() as http_mocker:
+        http_mocker.get(
+            HttpRequest(url="https://api.test.com/keys_replace/schema"),
+            HttpResponse(
+                body=json.dumps(
+                    {
+                        "fields": [
+                            {"name": "hs_v2_date_entered_customer", "type": "string"},
+                            {"name": "hs_v2_date_entered_lead_date", "type": "string"},
+                            {"name": "hs_lifecyclestage_lead_date", "type": "integer"},
+                        ]
+                    }
+                )
+            ),
+        )
+
+        actual_catalog = source.discover(logger=source.logger, config=_CONFIG)
+
+    assert len(actual_catalog.streams) == 1
+    assert actual_catalog.streams[0].json_schema["properties"] == expected_properties
+
+
 def test_dynamic_schema_loader_with_type_conditions():
     _MANIFEST_WITH_TYPE_CONDITIONS = deepcopy(_MANIFEST)
     _MANIFEST_WITH_TYPE_CONDITIONS["definitions"]["party_members_stream"]["schema_loader"][
diff --git a/unit_tests/sources/declarative/transformations/test_keys_replace_transformation.py b/unit_tests/sources/declarative/transformations/test_keys_replace_transformation.py
index 417d992afc..96c37a9b75 100644
--- a/unit_tests/sources/declarative/transformations/test_keys_replace_transformation.py
+++ b/unit_tests/sources/declarative/transformations/test_keys_replace_transformation.py
@@ -110,3 +110,250 @@ def test_transform(
         record=input_record, config=config, stream_state=stream_state, stream_slice=stream_slice
     )
     assert input_record == expected_record
+
+
+@pytest.mark.parametrize(
+    ["input_record", "config", "keys_replace_config", "expected_record"],
+    [
+        pytest.param(
+            {"a b": 1, "nested": {"c d": 2, "deeper": {"e f": 3}}},
+            {},
+            {"old": " ", "new": "_"},
+            {"a_b": 1, "nested": {"c_d": 2, "deeper": {"e_f": 3}}},
+            id="defaults keep literal, recursive, rename behavior",
+        ),
+        pytest.param(
+            {"properties": {"hs_v2_date_entered_poetry": 1827, "hs_date_entered_poetry": 9999}},
+            {},
+            {"old": "hs_v2_date_entered_", "new": "hs_date_entered_"},
+            {"properties": {"hs_date_entered_poetry": 9999}},
+            id="defaults keep last-write-wins collision behavior",
+        ),
+        pytest.param(
+            {"hs_v2_date_entered_customer": 1, "other": 2},
+            {},
+            {"old": "hs_v2_date_entered_(.*)", "new": r"hs_lifecyclestage_\1_date", "regex": True},
+            {"hs_lifecyclestage_customer_date": 1, "other": 2},
+            id="regex with numeric backreference",
+        ),
+        pytest.param(
+            {"hs_v2_date_entered_customer": 1},
+            {},
+            {
+                "old": "hs_v2_date_entered_(?P<stage>.*)",
+                "new": r"hs_lifecyclestage_\g<stage>_date",
+                "regex": True,
+            },
+            {"hs_lifecyclestage_customer_date": 1},
+            id="regex with named group backreference",
+        ),
+        pytest.param(
+            {"hs_v2_date_entered_customer": 1},
+            {},
+            {
+                "old": "hs_v2_date_entered_(.*)",
+                "new": r"hs_lifecyclestage_\g<1>_date",
+                "regex": True,
+            },
+            {"hs_lifecyclestage_customer_date": 1},
+            id="regex with g<1> backreference",
+        ),
+        pytest.param(
+            {"hs_v2_date_entered_customer": 1, "hs_v2_date_entered_lead_date": 2},
+            {},
+            {
+                "old": "hs_v2_date_entered_(.*?)(?:_date)?$",
+                "new": r"hs_lifecyclestage_\1_date",
+                "regex": True,
+            },
+            {"hs_lifecyclestage_customer_date": 1, "hs_lifecyclestage_lead_date": 2},
+            id="regex does not double append a suffix",
+        ),
+        pytest.param(
+            {"id_1": 1, "id_22": 2, "name": 3},
+            {},
+            {"old": r"_(\d+)$", "new": r"[\1]", "regex": True},
+            {"id[1]": 1, "id[22]": 2, "name": 3},
+            id="regex with backslash escapes in the pattern",
+        ),
+        pytest.param(
+            {"x(1)": 1},
+            {},
+            {"old": "(1)", "new": "one", "regex": True},
+            {"x(one)": 1},
+            id="plain regex is not literal-evaluated by interpolation",
+        ),
+        pytest.param(
+            {"prefix_a": 1, "prefix_b": 2},
+            {"source_prefix": "prefix_"},
+            {"old": "^{{ config['source_prefix'] }}(.*)", "new": r"\1", "regex": True},
+            {"a": 1, "b": 2},
+            id="interpolated regex keeps backreferences",
+        ),
+        pytest.param(
+            {"hs_v2_date_entered_x": 1, "unrelated": 2},
+            {},
+            {"old": "hs_v2_date_entered_", "new": "hs_date_entered_", "keep_original": True},
+            {"hs_v2_date_entered_x": 1, "hs_date_entered_x": 1, "unrelated": 2},
+            id="keep_original copies matching keys and does not duplicate others",
+        ),
+        pytest.param(
+            {"hs_v2_date_entered_x": 1, "hs_date_entered_x": 9999},
+            {},
+            {"old": "hs_v2_date_entered_", "new": "hs_date_entered_", "keep_original": True},
+            {"hs_v2_date_entered_x": 1, "hs_date_entered_x": 1},
+            id="keep_original overwrites an existing target regardless of key order",
+        ),
+        pytest.param(
+            {
+                "hs_v2_date_entered_absent": 1,
+                "hs_v2_date_entered_null": 2,
+                "hs_date_entered_null": None,
+                "hs_v2_date_entered_present": 3,
+                "hs_date_entered_present": 9999,
+            },
+            {},
+            {
+                "old": "hs_v2_date_entered_",
+                "new": "hs_date_entered_",
+                "keep_original": True,
+                "only_if_missing": True,
+            },
+            {
+                "hs_v2_date_entered_absent": 1,
+                "hs_date_entered_absent": 1,
+                "hs_v2_date_entered_null": 2,
+                "hs_date_entered_null": 2,
+                "hs_v2_date_entered_present": 3,
+                "hs_date_entered_present": 9999,
+            },
+            id="keep_original with only_if_missing for absent, null and present targets",
+        ),
+        pytest.param(
+            {
+                "hs_v2_date_entered_absent": 1,
+                "hs_v2_date_entered_null": 2,
+                "hs_date_entered_null": None,
+                "hs_v2_date_entered_present": 3,
+                "hs_date_entered_present": 9999,
+            },
+            {},
+            {"old": "hs_v2_date_entered_", "new": "hs_date_entered_", "only_if_missing": True},
+            {
+                "hs_date_entered_absent": 1,
+                "hs_date_entered_null": 2,
+                "hs_v2_date_entered_present": 3,
+                "hs_date_entered_present": 9999,
+            },
+            id="rename with only_if_missing keeps the original key when the target is present",
+        ),
+        pytest.param(
+            {"hs_date_entered_null": None, "hs_v2_date_entered_null": 2},
+            {},
+            {"old": "hs_v2_date_entered_", "new": "hs_date_entered_", "only_if_missing": True},
+            {"hs_date_entered_null": 2},
+            id="rename with only_if_missing fills a null target declared before the source key",
+        ),
+        pytest.param(
+            {"a b": 1, "properties": {"c d": 2, "nested": {"e f": 3}}},
+            {},
+            {"old": " ", "new": "_", "field_path": ["properties"]},
+            {"a b": 1, "properties": {"c_d": 2, "nested": {"e_f": 3}}},
+            id="field_path scopes the replacement and recurses within the target",
+        ),
+        pytest.param(
+            {"a b": 1, "attrs": {"c d": 2}},
+            {"target": "attrs"},
+            {"old": " ", "new": "_", "field_path": ["{{ config['target'] }}"]},
+            {"a b": 1, "attrs": {"c_d": 2}},
+            id="field_path is interpolated with config",
+        ),
+        pytest.param(
+            {"data": [{"attributes": {"c d": 1}}, {"attributes": {"e f": 2}}, {"other": 3}]},
+            {},
+            {"old": " ", "new": "_", "field_path": ["data", "*", "attributes"]},
+            {"data": [{"attributes": {"c_d": 1}}, {"attributes": {"e_f": 2}}, {"other": 3}]},
+            id="field_path wildcard applies to every matching object",
+        ),
+        pytest.param(
+            {"a b": 1},
+            {},
+            {"old": " ", "new": "_", "field_path": ["properties"]},
+            {"a b": 1},
+            id="field_path missing is a no-op",
+        ),
+        pytest.param(
+            {"a b": 1, "properties": "not a dict"},
+            {},
+            {"old": " ", "new": "_", "field_path": ["properties"]},
+            {"a b": 1, "properties": "not a dict"},
+            id="field_path pointing to a non object is a no-op",
+        ),
+        pytest.param(
+            {"properties": {"a b": {"c d": 1}}},
+            {},
+            {"old": " ", "new": "_", "field_path": ["properties"], "keep_original": True},
+            {"properties": {"a b": {"c d": 1, "c_d": 1}, "a_b": {"c d": 1, "c_d": 1}}},
+            id="keep_original applies recursively to nested objects",
+        ),
+    ],
+)
+def test_transform_with_options(input_record, config, keys_replace_config, expected_record):
+    KeysReplaceTransformation(**keys_replace_config, parameters={}).transform(
+        record=input_record, config=config
+    )
+    assert input_record == expected_record
+
+
+def test_keep_original_does_not_alias_nested_values():
+    record = {"a b": {"c": 1}}
+    KeysReplaceTransformation(old=" ", new="_", keep_original=True, parameters={}).transform(record)
+    record["a_b"]["c"] = 2
+    assert record["a b"] == {"c": 1}
+
+
+def test_new_keys_are_not_processed_again():
+    record = {"a": 1}
+    KeysReplaceTransformation(old="a", new="aa", keep_original=True, parameters={}).transform(
+        record
+    )
+    assert record == {"a": 1, "aa": 1}
+
+
+@pytest.mark.parametrize(
+    ["keys_replace_config", "expected_error"],
+    [
+        pytest.param(
+            {"old": "(unclosed", "new": "x"},
+            "KeysReplace `old` is not a valid regular expression",
+            id="invalid pattern",
+        ),
+        pytest.param(
+            {"old": "(a)", "new": r"\2"},
+            "KeysReplace `new` is not a valid replacement",
+            id="invalid group reference",
+        ),
+        pytest.param(
+            {"old": "(a)", "new": r"\g<missing>"},
+            "KeysReplace `new` is not a valid replacement",
+            id="unknown group name",
+        ),
+    ],
+)
+def test_invalid_regex_fails_at_construction(keys_replace_config, expected_error):
+    with pytest.raises(ValueError, match=expected_error):
+        KeysReplaceTransformation(**keys_replace_config, regex=True, parameters={})
+
+
+def test_invalid_interpolated_regex_fails_at_transform():
+    transformation = KeysReplaceTransformation(
+        old="{{ config['pattern'] }}", new="x", regex=True, parameters={}
+    )
+    with pytest.raises(ValueError, match="KeysReplace `old` is not a valid regular expression"):
+        transformation.transform({"a": 1}, config={"pattern": "(unclosed"})
+
+
+def test_literal_mode_does_not_validate_regex():
+    record = {"a(b": 1}
+    KeysReplaceTransformation(old="(", new="_", parameters={}).transform(record)
+    assert record == {"a_b": 1}