Repository: kedro-org/kedro

Issue:
Prioritise user created catch-all dataset factory pattern over `{default}` from runners
## Description
From slack conversation with @noklam https://linen-slack.kedro.org/t/16713675/i-am-trying-to-create-a-catch-all-patterns-using-the-dataset#65f8bddf-cb66-4b35-8f5d-c0850669b907

The catch-all factory pattern added by the user in their catalog does not always override the default dataset set by the runner - 
This works as expected ✅ -
```
"{catch_all}":
  type: pickle.PickleDataset
  filepath: data/06_models/{catch_all}.pickle
```
However, this pattern doesn't ❌  - 
```
"{nok}":
  type: pickle.PickleDataset
  filepath: data/06_models/{nok}.pickle
```

## Context
This behaviour was introduced in https://github.com/kedro-org/kedro/pull/3332 and was intentional and [is documented](https://docs.kedro.org/en/stable/data/kedro_dataset_factories.html#how-to-override-the-default-dataset-creation-with-dataset-factories), however, it is a little bit strange since for a user defined "catch-all" pattern to be used instead of the default dataset defined by the runner, which is `{default}` has to be ranked higher in terms of priority (alphabetically) to be picked up. That limits the names of the catch all patterns (should begin with a/b/c etc). 


## Expected Result
- Should always use the catch all pattern defined by the user (if there is one) instead of the default provided by the runner
- `{default}` can still be a protected catch-all pattern name that can only be used by the runner
- We should add a regression test for this behaviour. The catch-all patterns behave as expected in 0.18.14 but not with 0.19.x



Discussion:
noklam: >  [Question for reviewers]: We should probably error out if the user has more than one catch-all patterns in this case, does that make sense?

Yes
noklam: Just dumping my thought here in case I am blocking the PRs. I have some questions about the `shallow_copy` approach, but that was done long before this PR so I don't want to expand the discussion too much.

I think:
- There should be only 1 default dataset, validation should happen when `DataCatalog` is initialised. (I think this is consistent as the `shallow_copy` will create a new `DataCatalog`, this is where I find the `shallow_copy` weird as it [does not seem to be a reference but completely new object](https://realpython.com/copying-python-objects/#:~:text=A%20shallow%20copy%20means%20constructing,of%20the%20child%20objects%20themselves.)
- multiple dataset pattern by Runner is fine. As long as only one of them is default datasets. For example, you could have different default for different namespaces which control by runner.

The most important question here is "should runner dataset patterns override `DataCatalog`"? In framework setting, the way to override patterns is `catalog.yml`. However, the order of how framework work is different. 
1. `DataCatalog` read `catalog.yml` first 
2.  Runner override default patterns (memory dataset)

This suggests that one should only use either `catalog.yml` or `runner`. Should it fails or just ignore the dataset pattern from runner? For me, I think I will go for raising an error instead of ignoring the pattern sliently.

Patch:
diff --git a/RELEASE.md b/RELEASE.md
index 34c23572..620abccf 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -3,6 +3,7 @@
 ## Major features and improvements
 
 ## Bug fixes and other changes
+* User defined catch-all dataset factory patterns now override the default pattern provided by the runner.
 
 ## Breaking changes to the API
 
diff --git a/docs/source/data/kedro_dataset_factories.md b/docs/source/data/kedro_dataset_factories.md
index fea61cad..cd591e35 100644
--- a/docs/source/data/kedro_dataset_factories.md
+++ b/docs/source/data/kedro_dataset_factories.md
@@ -223,19 +223,14 @@ The matches are ranked according to the following criteria:
 You can use dataset factories to define a catch-all pattern which will overwrite the default [`MemoryDataset`](/api/kedro.io.MemoryDataset) creation.
 
 ```yaml
-"{a_default_dataset}":
+"{default_dataset}":
   type: pandas.CSVDataset
-  filepath: data/{a_default_dataset}.csv
+  filepath: data/{default_dataset}.csv
 
 ```
 Kedro will now treat all the datasets mentioned in your project's pipelines that do not appear as specific patterns or explicit entries in your catalog
 as `pandas.CSVDataset`.
 
-```{note}
-Under the hood Kedro uses the pattern name "{default}" to generate the default datasets set in the runners. If you want to overwrite this pattern you should make sure you choose a name that comes
-before "default" in the alphabet for it to be resolved first.
-```
-
 ## CLI commands for dataset factories
 
 To manage your dataset factories, two new commands have been added to the Kedro CLI: `kedro catalog rank` (0.18.12) and `kedro catalog resolve` (0.18.13).
@@ -322,9 +317,9 @@ shuttles:
   type: pandas.ParquetDataset
   filepath: data/02_intermediate/preprocessed_{name}.pq
 
-"{a_default}":
+"{default}":
   type: pandas.ParquetDataset
-  filepath: data/03_primary/{a_default}.pq
+  filepath: data/03_primary/{default}.pq
 ```
 </details>
 
diff --git a/kedro/framework/cli/catalog.py b/kedro/framework/cli/catalog.py
index a07b5d4b..a0f6b413 100644
--- a/kedro/framework/cli/catalog.py
+++ b/kedro/framework/cli/catalog.py
@@ -87,10 +87,12 @@ def list_datasets(metadata: ProjectMetadata, pipeline: str, env: str) -> None:
         for ds_name in default_ds:
             matched_pattern = data_catalog._match_pattern(
                 data_catalog._dataset_patterns, ds_name
-            )
+            ) or data_catalog._match_pattern(data_catalog._default_pattern, ds_name)
             if matched_pattern:
                 ds_config_copy = copy.deepcopy(
-                    data_catalog._dataset_patterns[matched_pattern]
+                    data_catalog._dataset_patterns.get(matched_pattern)
+                    or data_catalog._default_pattern.get(matched_pattern)
+                    or {}
                 )
 
                 ds_config = data_catalog._resolve_config(
@@ -215,7 +217,10 @@ def rank_catalog_factories(metadata: ProjectMetadata, env: str) -> None:
     session = _create_session(metadata.package_name, env=env)
     context = session.load_context()
 
-    catalog_factories = context.catalog._dataset_patterns
+    catalog_factories = {
+        **context.catalog._dataset_patterns,
+        **context.catalog._default_pattern,
+    }
     if catalog_factories:
         click.echo(yaml.dump(list(catalog_factories.keys())))
     else:
@@ -259,10 +264,12 @@ def resolve_patterns(metadata: ProjectMetadata, env: str) -> None:
 
         matched_pattern = data_catalog._match_pattern(
             data_catalog._dataset_patterns, ds_name
-        )
+        ) or data_catalog._match_pattern(data_catalog._default_pattern, ds_name)
         if matched_pattern:
             ds_config_copy = copy.deepcopy(
-                data_catalog._dataset_patterns[matched_pattern]
+                data_catalog._dataset_patterns.get(matched_pattern)
+                or data_catalog._default_pattern.get(matched_pattern)
+                or {}
             )
 
             ds_config = data_catalog._resolve_config(
diff --git a/kedro/io/data_catalog.py b/kedro/io/data_catalog.py
index 411cf14e..ea0fb12f 100644
--- a/kedro/io/data_catalog.py
+++ b/kedro/io/data_catalog.py
@@ -144,6 +144,7 @@ class DataCatalog:
         dataset_patterns: Patterns | None = None,
         load_versions: dict[str, str] | None = None,
         save_version: str | None = None,
+        default_pattern: Patterns | None = None,
     ) -> None:
         """``DataCatalog`` stores instances of ``AbstractDataset``
         implementations to provide ``load`` and ``save`` capabilities from
@@ -172,6 +173,8 @@ class DataCatalog:
                 case-insensitive string that conforms with operating system
                 filename limitations, b) always return the latest version when
                 sorted in lexicographical order.
+            default_pattern: A dictionary of the default catch-all pattern that overrides the default
+                pattern provided through the runners.
 
         Example:
         ::
@@ -190,6 +193,7 @@ class DataCatalog:
         self._dataset_patterns = dataset_patterns or {}
         self._load_versions = load_versions or {}
         self._save_version = save_version
+        self._default_pattern = default_pattern or {}
 
         if feed_dict:
             self.add_feed_dict(feed_dict)
@@ -281,6 +285,7 @@ class DataCatalog:
         credentials = copy.deepcopy(credentials) or {}
         save_version = save_version or generate_timestamp()
         load_versions = copy.deepcopy(load_versions) or {}
+        user_default = {}
 
         for ds_name, ds_config in catalog.items():
             ds_config = _resolve_credentials(  # noqa: PLW2901
@@ -295,6 +300,12 @@ class DataCatalog:
                     ds_name, ds_config, load_versions.get(ds_name), save_version
                 )
         sorted_patterns = cls._sort_patterns(dataset_patterns)
+        if sorted_patterns:
+            # If the last pattern is a catch-all pattern, pop it and set it as the default
+            if cls._specificity(list(sorted_patterns.keys())[-1]) == 0:
+                last_pattern = sorted_patterns.popitem()
+                user_default = {last_pattern[0]: last_pattern[1]}
+
         missing_keys = [
             key
             for key in load_versions.keys()
@@ -311,6 +322,7 @@ class DataCatalog:
             dataset_patterns=sorted_patterns,
             load_versions=load_versions,
             save_version=save_version,
+            default_pattern=user_default,
         )
 
     @staticmethod
@@ -346,6 +358,13 @@ class DataCatalog:
                 pattern,
             ),
         )
+        catch_all = [
+            pattern for pattern in sorted_keys if cls._specificity(pattern) == 0
+        ]
+        if len(catch_all) > 1:
+            raise DatasetError(
+                f"Multiple catch-all patterns found in the catalog: {', '.join(catch_all)}. Only one catch-all pattern is allowed, remove the extras."
+            )
         return {key: dataset_patterns[key] for key in sorted_keys}
 
     @staticmethod
@@ -369,11 +388,17 @@ class DataCatalog:
         version: Version | None = None,
         suggest: bool = True,
     ) -> AbstractDataset:
-        matched_pattern = self._match_pattern(self._dataset_patterns, dataset_name)
+        matched_pattern = self._match_pattern(
+            self._dataset_patterns, dataset_name
+        ) or self._match_pattern(self._default_pattern, dataset_name)
         if dataset_name not in self._datasets and matched_pattern:
             # If the dataset is a patterned dataset, materialise it and add it to
             # the catalog
-            config_copy = copy.deepcopy(self._dataset_patterns[matched_pattern])
+            config_copy = copy.deepcopy(
+                self._dataset_patterns.get(matched_pattern)
+                or self._default_pattern.get(matched_pattern)
+                or {}
+            )
             dataset_config = self._resolve_config(
                 dataset_name, matched_pattern, config_copy
             )
@@ -385,7 +410,7 @@ class DataCatalog:
             )
             if (
                 self._specificity(matched_pattern) == 0
-                and matched_pattern != "{default}"
+                and matched_pattern in self._default_pattern
             ):
                 self._logger.warning(
                     "Config from the dataset factory pattern '%s' in the catalog will be used to "
@@ -721,7 +746,7 @@ class DataCatalog:
         Returns:
             Copy of the current object.
         """
-        if extra_dataset_patterns:
+        if not self._default_pattern and extra_dataset_patterns:
             unsorted_dataset_patterns = {
                 **self._dataset_patterns,
                 **extra_dataset_patterns,
@@ -734,6 +759,7 @@ class DataCatalog:
             dataset_patterns=dataset_patterns,
             load_versions=self._load_versions,
             save_version=self._save_version,
+            default_pattern=self._default_pattern,
         )
 
     def __eq__(self, other) -> bool:  # type: ignore[no-untyped-def]


Test patch:
diff --git a/tests/io/test_data_catalog.py b/tests/io/test_data_catalog.py
index 9f424c04..d833aa73 100644
--- a/tests/io/test_data_catalog.py
+++ b/tests/io/test_data_catalog.py
@@ -151,10 +151,6 @@ def config_with_dataset_factories_bad_pattern(config_with_dataset_factories):
 def config_with_dataset_factories_only_patterns():
     return {
         "catalog": {
-            "{default}": {
-                "type": "pandas.CSVDataset",
-                "filepath": "data/01_raw/{default}.csv",
-            },
             "{namespace}_{dataset}": {
                 "type": "pandas.CSVDataset",
                 "filepath": "data/01_raw/{namespace}_{dataset}.pq",
@@ -175,6 +171,14 @@ def config_with_dataset_factories_only_patterns():
     }
 
 
+@pytest.fixture
+def config_with_dataset_factories_only_patterns_no_default(
+    config_with_dataset_factories_only_patterns,
+):
+    del config_with_dataset_factories_only_patterns["catalog"]["{user_default}"]
+    return config_with_dataset_factories_only_patterns
+
+
 @pytest.fixture
 def dataset(filepath):
     return CSVDataset(filepath=filepath, save_args={"index": False})
@@ -841,15 +845,37 @@ class TestDataCatalogDatasetFactories:
             "{country}_companies",
             "{namespace}_{dataset}",
             "{dataset}s",
-            "{default}",
             "{user_default}",
         ]
-        assert list(catalog._dataset_patterns.keys()) == sorted_keys_expected
+        assert (
+            list(catalog._dataset_patterns.keys())
+            + list(catalog._default_pattern.keys())
+            == sorted_keys_expected
+        )
 
-    def test_sorting_order_with_default_and_other_dataset_through_extra_pattern(
-        self, config_with_dataset_factories_only_patterns
+    def test_multiple_catch_all_patterns_not_allowed(
+        self, config_with_dataset_factories
     ):
-        """Check that the sorted order of the patterns is correct according to parsing rules when a default dataset is added through extra patterns (this would happen via the runner)."""
+        """Check that multiple catch-all patterns are not allowed"""
+        config_with_dataset_factories["catalog"]["{default1}"] = {
+            "filepath": "data/01_raw/{default1}.csv",
+            "type": "pandas.CSVDataset",
+        }
+        config_with_dataset_factories["catalog"]["{default2}"] = {
+            "filepath": "data/01_raw/{default2}.xlsx",
+            "type": "pandas.ExcelDataset",
+        }
+
+        with pytest.raises(
+            DatasetError, match="Multiple catch-all patterns found in the catalog"
+        ):
+            DataCatalog.from_config(**config_with_dataset_factories)
+
+    def test_sorting_order_with_other_dataset_through_extra_pattern(
+        self, config_with_dataset_factories_only_patterns_no_default
+    ):
+        """Check that the sorted order of the patterns is correct according to parsing rules when a default dataset
+        is added through extra patterns (this would happen via the runner) and user default is not present"""
         extra_dataset_patterns = {
             "{default}": {"type": "MemoryDataset"},
             "{another}#csv": {
@@ -857,7 +883,9 @@ class TestDataCatalogDatasetFactories:
                 "filepath": "data/{another}.csv",
             },
         }
-        catalog = DataCatalog.from_config(**config_with_dataset_factories_only_patterns)
+        catalog = DataCatalog.from_config(
+            **config_with_dataset_factories_only_patterns_no_default
+        )
         catalog_with_default = catalog.shallow_copy(
             extra_dataset_patterns=extra_dataset_patterns
         )
@@ -867,38 +895,13 @@ class TestDataCatalogDatasetFactories:
             "{namespace}_{dataset}",
             "{dataset}s",
             "{default}",
-            "{user_default}",
         ]
         assert (
             list(catalog_with_default._dataset_patterns.keys()) == sorted_keys_expected
         )
 
-    def test_runner_default_overwrites_user_default(
-        self, config_with_dataset_factories_only_patterns
-    ):
-        """Check that the runner default overwrites the user default."""
-        catalog = DataCatalog.from_config(**config_with_dataset_factories_only_patterns)
-        assert catalog._dataset_patterns["{default}"] == {
-            "filepath": "data/01_raw/{default}.csv",
-            "type": "pandas.CSVDataset",
-        }
-
-        extra_dataset_patterns = {
-            "{default}": {"type": "MemoryDataset"},
-            "{another}#csv": {
-                "type": "pandas.CSVDataset",
-                "filepath": "data/{another}.csv",
-            },
-        }
-        catalog_with_runner_default = catalog.shallow_copy(
-            extra_dataset_patterns=extra_dataset_patterns
-        )
-        assert catalog_with_runner_default._dataset_patterns["{default}"] == {
-            "type": "MemoryDataset"
-        }
-
-    def test_user_default_overwrites_runner_default_alphabetically(self):
-        """Check that the runner default overwrites the user default if earlier in alphabet."""
+    def test_user_default_overwrites_runner_default(self):
+        """Check that the user default overwrites the runner default when both are present"""
         catalog_config = {
             "{dataset}s": {
                 "type": "pandas.CSVDataset",
@@ -921,13 +924,13 @@ class TestDataCatalogDatasetFactories:
             extra_dataset_patterns=extra_dataset_patterns
         )
         sorted_keys_expected = [
-            "{another}#csv",
             "{dataset}s",
             "{a_default}",
-            "{default}",
         ]
+        assert "{a_default}" in catalog_with_runner_default._default_pattern
         assert (
             list(catalog_with_runner_default._dataset_patterns.keys())
+            + list(catalog_with_runner_default._default_pattern.keys())
             == sorted_keys_expected
         )