Repository: kedro-org/kedro

Issue:
Use soft-merge for certain configuration files for `OmegaConfLoader`
## Description
At the moment all configuration files are merged in a destructive way. This means all fields/keys are overwritten when two files are merged e.g.:

Base catalog file:
```yml
regressor:
  type: pickle.PickleDataSet
  filepath: data/06_models/regressor.pickle
  versioned: true
```

Overriding catalog file:
```yml
regressor:
  type: pandas.CSVDataSet
```

Will result in:
```yml
regressor:
  type: pandas.CSVDataSet
```

With catalog and logging files this behaviour is desirable because if you'd like to change a file type, you don't want any fields to remain that belonged to the other file type (e.g. `plotly_args:` for the `PlotlyDataSet`). However, with parameter or other config files (e.g. for mlflow) it might be better to merge them in a soft way so you can do something like: 

Base `mlflow.yml` file:
```yml
tracking:
  disable_tracking:
    pipelines:
    - on_exit_notification

  experiment:
    name: name-of-local-experiment

  params:
    long_params_strategy: tag
```

Overriding `mlflow.yml` file:
```yml
tracking:
  experiment:
    name: name-of-prod-experiment
```

Results in:
```yml
tracking:
  disable_tracking:
    pipelines:
    - on_exit_notification
  experiment:
    name: name-of-prod-experiment
  params:
    long_params_strategy: tag
```

This can be achieved my using the OmegaConf merge: https://omegaconf.readthedocs.io/en/2.2_branch/usage.html#merging-configurations 

## Context
https://github.com/kedro-org/kedro/issues/1868#issuecomment-1285020986

## Possible Implementation
Add another setting to `OmegaConfLoader` where the user can specify which files to merge in a destructive way and which one with the regular OmegaConf merge.

* Check if it's possible to remove keys with the `OmegaConf` merge.

## Possible Alternative
Merge `catalog` files in a destructive way and all other in a soft way. This would be a breaking change. 


Discussion:
merelcht: > Nice use of OmegaConf.merge. I was not aware that the merging strategy is for merging between two environments only (not for multiple config in same env), maybe we could make this more explicit in docs.

I added more explanation to the docs, let me know if that's better!

Patch:
diff --git a/RELEASE.md b/RELEASE.md
index d1e99317..f444a298 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -4,6 +4,7 @@
 * Dropped Python 3.7 support.
 * Introduced add-ons to the `kedro new` CLI flow.
 * The new spaceflights starters, `spaceflights-pandas`, `spaceflights-pandas-viz`, `spaceflights-pyspark`, and `spaceflights-pyspark-viz` can be used with the `kedro new` command with the `--starter` flag.
+* Added the functionality to choose a merging strategy for config files loaded with `OmegaConfigLoader`.
 
 ## Bug fixes and other changes
 * Added a new field `add-ons` to `pyproject.toml` when a project is created.
diff --git a/docs/source/configuration/advanced_configuration.md b/docs/source/configuration/advanced_configuration.md
index 27e7ebc8..a67f194d 100644
--- a/docs/source/configuration/advanced_configuration.md
+++ b/docs/source/configuration/advanced_configuration.md
@@ -15,6 +15,7 @@ This page also contains a set of guidance for advanced configuration requirement
 * [How to override configuration with runtime parameters with the `OmegaConfigLoader`](#how-to-override-configuration-with-runtime-parameters-with-the-omegaconfigloader)
 * [How to use resolvers in the `OmegaConfigLoader`](#how-to-use-resolvers-in-the-omegaconfigloader)
 * [How to load credentials through environment variables with `OmegaConfigLoader`](#how-to-load-credentials-through-environment-variables)
+* [How to change the merge strategy used by `OmegaConfigLoader`](#how-to-change-the-merge-strategy-used-by-omegaconfigloader)
 
 ## OmegaConfigLoader
 
@@ -414,3 +415,24 @@ dev_s3:
 ```{note}
 Note that you can only use the resolver in `credentials.yml` and not in catalog or parameter files. This is because we do not encourage the usage of environment variables for anything other than credentials.
 ```
+
+### How to change the merge strategy used by `OmegaConfigLoader`
+By default, `OmegaConfigLoader` merges configuration [in different environments](configuration_basics.md#configuration-environments) in a destructive way. This means that whatever configuration resides in your overriding environment (`local` by default) takes precedence when the same top-level key is present in the base and overriding environment. Any configuration for that key **besides that given in the overriding environment** is discarded.
+You can change the merge strategy for each configuration type in your project's `src/<package_name>/settings.py`. The accepted merging strategies are `soft` and `destructive`.
+
+```python
+from kedro.config import OmegaConfigLoader
+
+CONFIG_LOADER_CLASS = OmegaConfigLoader
+
+CONFIG_LOADER_ARGS = {
+    "merge_strategy": {
+        "parameters": "soft",
+        "spark": "destructive",
+        "mlflow": "soft",
+    }
+}
+```
+
+If no merge strategy is defined, the default destructive strategy will be applied. Note that this merge strategy setting only applies to configuration files in **different** environments.
+When files are part of the same environment, they are always merged in a soft way. An error is thrown when files in the same environment contain the same top-level keys.
diff --git a/kedro/config/omegaconf_config.py b/kedro/config/omegaconf_config.py
index af26b654..6fb281f5 100644
--- a/kedro/config/omegaconf_config.py
+++ b/kedro/config/omegaconf_config.py
@@ -83,6 +83,7 @@ class OmegaConfigLoader(AbstractConfigLoader):
         base_env: str = "base",
         default_run_env: str = "local",
         custom_resolvers: dict[str, Callable] = None,
+        merge_strategy: dict[str, str] = None,
     ):
         """Instantiates a ``OmegaConfigLoader``.
 
@@ -103,6 +104,7 @@ class OmegaConfigLoader(AbstractConfigLoader):
         """
         self.base_env = base_env
         self.default_run_env = default_run_env
+        self.merge_strategy = merge_strategy or {}
 
         self.config_patterns = {
             "catalog": ["catalog*", "catalog*/**", "**/catalog*"],
@@ -224,17 +226,16 @@ class OmegaConfigLoader(AbstractConfigLoader):
             else:
                 raise exc
 
-        # Destructively merge the two env dirs. The chosen env will override base.
-        common_keys = config.keys() & env_config.keys()
-        if common_keys:
-            sorted_keys = ", ".join(sorted(common_keys))
-            msg = (
-                "Config from path '%s' will override the following "
-                "existing top-level config keys: %s"
+        merging_strategy = self.merge_strategy.get(key)
+        if merging_strategy == "soft":
+            resulting_config = self._soft_merge(config, env_config)
+        elif merging_strategy == "destructive" or not merging_strategy:
+            resulting_config = self._destructive_merge(config, env_config, env_path)
+        else:
+            raise ValueError(
+                f"Merging strategy {merging_strategy} not supported. The accepted merging "
+                f"strategies are `soft` and `destructive`."
             )
-            _config_logger.debug(msg, env_path, sorted_keys)
-
-        config.update(env_config)
 
         if not processed_files and key != "globals":
             raise MissingConfigException(
@@ -242,7 +243,7 @@ class OmegaConfigLoader(AbstractConfigLoader):
                 f" the glob pattern(s): {[*self.config_patterns[key]]}"
             )
 
-        return config
+        return resulting_config
 
     def __repr__(self):  # pragma: no cover
         return (
@@ -445,6 +446,26 @@ class OmegaConfigLoader(AbstractConfigLoader):
         else:
             OmegaConf.resolve(config)
 
+    @staticmethod
+    def _destructive_merge(config, env_config, env_path):
+        # Destructively merge the two env dirs. The chosen env will override base.
+        common_keys = config.keys() & env_config.keys()
+        if common_keys:
+            sorted_keys = ", ".join(sorted(common_keys))
+            msg = (
+                "Config from path '%s' will override the following "
+                "existing top-level config keys: %s"
+            )
+            _config_logger.debug(msg, env_path, sorted_keys)
+
+        config.update(env_config)
+        return config
+
+    @staticmethod
+    def _soft_merge(config, env_config):
+        # Soft merge the two env dirs. The chosen env will override base if keys clash.
+        return OmegaConf.to_container(OmegaConf.merge(config, env_config))
+
     def _is_hidden(self, path: str):
         """Check if path contains any hidden directory or is a hidden file"""
         path = Path(path)


Test patch:
diff --git a/tests/config/test_omegaconf_config.py b/tests/config/test_omegaconf_config.py
index ef6b8efb..3304175e 100644
--- a/tests/config/test_omegaconf_config.py
+++ b/tests/config/test_omegaconf_config.py
@@ -121,6 +121,31 @@ def proj_credentials_env_variable(tmp_path):
     )
 
 
+@pytest.fixture
+def mlflow_config(tmp_path):
+    base_mlflow = tmp_path / _BASE_ENV / "mlflow.yml"
+    base_config = {
+        "tracking": {
+            "disable_tracking": {"pipelines": "[on_exit_notification]"},
+            "experiment": {
+                "name": "name-of-local-experiment",
+            },
+            "params": {"long_params_strategy": "tag"},
+        }
+    }
+    local_mlflow = tmp_path / _DEFAULT_RUN_ENV / "mlflow.yml"
+    local_config = {
+        "tracking": {
+            "experiment": {
+                "name": "name-of-prod-experiment",
+            },
+        }
+    }
+
+    _write_yaml(base_mlflow, base_config)
+    _write_yaml(local_mlflow, local_config)
+
+
 use_config_dir = pytest.mark.usefixtures("create_config_dir")
 use_proj_catalog = pytest.mark.usefixtures("proj_catalog")
 use_credentials_env_variable_yml = pytest.mark.usefixtures(
@@ -417,20 +442,29 @@ class TestOmegaConfigLoader:
             "**/params*",
         ]
 
-    def test_destructive_merging_strategy(self, tmp_path):
+    def test_default_destructive_merging_strategy(self, tmp_path, mlflow_config):
         mlflow_patterns = {"mlflow": ["mlflow*", "mlflow*/**", "**/mlflow*"]}
-        base_mlflow = tmp_path / _BASE_ENV / "mlflow.yml"
-        base_config = {
+        conf = OmegaConfigLoader(str(tmp_path), config_patterns=mlflow_patterns)[
+            "mlflow"
+        ]
+
+        assert conf == {
             "tracking": {
-                "disable_tracking": {"pipelines": "[on_exit_notification]"},
                 "experiment": {
-                    "name": "name-of-local-experiment",
+                    "name": "name-of-prod-experiment",
                 },
-                "params": {"long_params_strategy": "tag"},
             }
         }
-        local_mlflow = tmp_path / _DEFAULT_RUN_ENV / "mlflow.yml"
-        local_config = {
+
+    def test_destructive_merging_strategy(self, tmp_path, mlflow_config):
+        mlflow_patterns = {"mlflow": ["mlflow*", "mlflow*/**", "**/mlflow*"]}
+        conf = OmegaConfigLoader(
+            str(tmp_path),
+            config_patterns=mlflow_patterns,
+            merge_strategy={"mlflow": "destructive"},
+        )["mlflow"]
+
+        assert conf == {
             "tracking": {
                 "experiment": {
                     "name": "name-of-prod-experiment",
@@ -438,21 +472,33 @@ class TestOmegaConfigLoader:
             }
         }
 
-        _write_yaml(base_mlflow, base_config)
-        _write_yaml(local_mlflow, local_config)
-
-        conf = OmegaConfigLoader(str(tmp_path), config_patterns=mlflow_patterns)[
-            "mlflow"
-        ]
+    def test_soft_merging_strategy(self, tmp_path, mlflow_config):
+        mlflow_patterns = {"mlflow": ["mlflow*", "mlflow*/**", "**/mlflow*"]}
+        conf = OmegaConfigLoader(
+            str(tmp_path),
+            config_patterns=mlflow_patterns,
+            merge_strategy={"mlflow": "soft"},
+        )["mlflow"]
 
         assert conf == {
             "tracking": {
+                "disable_tracking": {"pipelines": "[on_exit_notification]"},
                 "experiment": {
                     "name": "name-of-prod-experiment",
                 },
+                "params": {"long_params_strategy": "tag"},
             }
         }
 
+    def test_unsupported_merge_strategy(self, tmp_path, mlflow_config):
+        mlflow_patterns = {"mlflow": ["mlflow*", "mlflow*/**", "**/mlflow*"]}
+        with pytest.raises(ValueError):
+            OmegaConfigLoader(
+                str(tmp_path),
+                config_patterns=mlflow_patterns,
+                merge_strategy={"mlflow": "hard"},
+            )["mlflow"]
+
     @use_config_dir
     def test_adding_extra_keys_to_confloader(self, tmp_path):
         """Make sure extra keys can be added directly to the config loader instance."""