Repository: alan-turing-institute/sktime (python)

Issue:
[BUG] IndexError when selecting `return_X_y=False` and `return_type` is not referencing to a pandas dataframe.
**Describe the bug**
<!--
A clear and concise description of what the bug is.
-->
As per the discussion in #2697, it was [found](https://github.com/alan-turing-institute/sktime/issues/2697#issuecomment-1196290311) that `_load_provided_dataset()` must throw an error if `return_type` is not a reference to `pd.DataFrame` and `return_X_y=False`. This happens because of code appends the response variable into predictors under the column name 'class_val'.

**To Reproduce**
<!--
Add a Minimal, Complete, and Verifiable example (for more details, see e.g. https://stackoverflow.com/help/mcve

If the code is too long, feel free to put it in a public gist and link it in the issue: https://gist.github.com
-->

```python
import sktime
from sktime.datasets._data_io import _load_provided_dataset

def test_load_provided_dataset():
    name = "UnitTest"                                    # referencing to dataset available for testing in the directory 'datasets/data/'
    X, y = _load_provided_dataset(name, return_type='np2D', return_X_y=False)
    print(X.shape, y.shape)

if __name__ == "__main__":
    print("Staring program")
    test_load_provided_dataset()
    print("Done")
```

**Output of above code**
```
PS D:\Projects\sktime\sktime\sktime\datasets\tests> D:/Anaconda/envs/sktime-dev/python temp_test_data_io.py
Staring program
Traceback (most recent call last):
  File "temp_test_data_io.py", line 11, in <module>
    test_load_provided_dataset()
  File "temp_test_data_io.py", line 6, in test_load_provided_dataset
    X, y = _load_provided_dataset(name, return_type='np2D', return_X_y=False)
  File "d:\projects\sktime\sktime\sktime\datasets\_data_io.py", line 224, in _load_provided_dataset
    X["class_val"] = pd.Series(y)
IndexError: only integers, slices (`:`), ellipsis (`...`), numpy.newaxis (`None`) and integer or boolean arrays are valid indices
```

**Expected behavior**
<!--
A clear and concise description of what you expected to happen.
-->
This clearly is an issue because we are not checking the instance type of X, which could be `np.array` or `pd.DataFrame`.

**Solutions(?)**
* _Solution 1: Check instance and append in the appropriate axis_
  Consider the following cases 
  - Univariate, 2D, equal length data (`X=np.random.randn(20, 24), y=np.random.randn(20,)`) then `output.shape=(20,25)`
  - Univariate 3D, equal length data (`X=np.random.randn(20,1,24), y=np.random.randn(20,)`) then `output.shape=(20,1,25)`
  - Multivariate, equal length data (`X=np.random.randn(20,6,24), y=np.random.randn(20,)`) then `output.shape=(20,6,25)`; requires tiling labels which might not be desirable.
 
  I think it should also work for unequal length data but I need confirmation. The only drawback of this solution apart from the repetition of y in the multivariate case is that it might be less intuitive.

* _Solution 2: Force convert the data into pandas dataframe and give a warning about the conversion._

**Final Comments**
- There are more solutions to this but I think these two are the most straightforward. I also have the code ready for the first solution and will send a pull request if that is chosen, otherwise I'll take a day's time to send a pull request if we move forward with any other solution.
- I can also add unit tests regarding the same, I don't know if having tests for private methods are a thing or not, but I think it won't hurt having more tests.

**Versions**
<details>

<!--
Please run the following code snippet and paste the output here:
 
from sktime import show_versions; show_versions()
-->
```
D:\Anaconda\envs\sktime-dev\lib\site-packages\_distutils_hack\__init__.py:33: UserWarning: Setuptools is replacing distutils.
  warnings.warn("Setuptools is replacing distutils.")

System:
    python: 3.8.13 | packaged by conda-forge | (default, Mar 25 2022, 05:59:00) [MSC v.1929 64 bit (AMD64)]
executable: D:\Anaconda\envs\sktime-dev\python.exe
   machine: Windows-10-10.0.22000-SP0

Python dependencies:
          pip: 22.1.2
   setuptools: 63.1.0
      sklearn: 1.1.1
       sktime: 0.13.0
  statsmodels: 0.13.2
        numpy: 1.22.4
        scipy: 1.8.1
       pandas: 1.4.3
   matplotlib: 3.5.2
       joblib: 1.1.0
        numba: 0.55.2
     pmdarima: None
      tsfresh: None
```
</details>

<!-- Thanks for contributing! -->


Pull request description:
[BUG] fix failure of some conversions in `_load_provided_dataset()`
<!--
Thanks for contributing a pull request! Please ensure you have taken a look
at our contribution guide: https://github.com/alan-turing-institute/sktime/blob/main/CONTRIBUTING.md
-->

#### Reference Issues/PRs
<!--
Example: Fixes #1234. See also #3456.

Please use keywords (e.g., Fixes) to create link to the issues or pull requests
you resolved, so that they will automatically be closed when your pull request
is merged. See https://github.com/blog/1506-closing-issues-via-pull-requests
-->
Fixes #3169


#### What does this implement/fix? Explain your changes.
<!--
A clear and concise description of what you have implemented.
-->
- Modify `_load_provided_dataset()` under `datasets/_data_io.py` to inhibit failures from data-type conversions.
- Add `return_type` parameter to `load_UCR_UEA_dataset()` under `datasets/_single_problem_loaders.py` giving even more flexibility.


#### PR checklist
<!--
Please go through the checklist below. Please feel free to remove points if they are not applicable.
-->

##### For all contributions
- [x] I've added unit tests and made sure they pass locally.
- [x] The PR title starts with either [ENH], [MNT], [DOC], or [BUG] indicating whether the PR topic is related to enhancement, maintenance, documentation, or bug.


<!--
Thanks for contributing!
-->


Patch:
diff --git a/.all-contributorsrc b/.all-contributorsrc
index b641b5a29..b4cc47ab2 100644
--- a/.all-contributorsrc
+++ b/.all-contributorsrc
@@ -589,7 +589,9 @@
       "avatar_url": "https://avatars.githubusercontent.com/u/54197164?v=4",
       "profile": "https://github.com/achieveordie",
       "contributions": [
-        "code"
+        "bug",
+        "code",
+        "doc"
       ]
     },
     {
diff --git a/sktime/datasets/_data_io.py b/sktime/datasets/_data_io.py
index 0d82a1c50..4106cf731 100644
--- a/sktime/datasets/_data_io.py
+++ b/sktime/datasets/_data_io.py
@@ -44,6 +44,17 @@ DIRNAME = "data"
 MODULE = os.path.dirname(__file__)
 
 
+# Return appropriate return_type in case an alias was used
+def _alias_mtype_check(return_type):
+    if return_type is None:
+        return_type = "nested_univ"
+    if return_type in ["numpy2d", "numpy2D", "np2d", "np2D"]:
+        return_type = "numpyflat"
+    if return_type in ["numpy3d", "np3d", "np3D"]:
+        return_type = "numpy3D"
+    return return_type
+
+
 # time series classification data sets
 def _download_and_extract(url, extract_path=None):
     """
@@ -182,7 +193,7 @@ def _load_provided_dataset(
         name : string, file name
         split : string, default = None, or one of "TRAIN" or "TEST".
         return_X_y : default = True, if true, returns X and y separately.
-        return_type : default = None,
+        return_type : string or None, default = None.
         local_module: default = os.path.dirname(__file__),
         local_dirname: default = "data"
     """
@@ -192,36 +203,31 @@ def _load_provided_dataset(
     if split in ("TRAIN", "TEST"):
         fname = name + "_" + split + ".ts"
         abspath = os.path.join(local_module, local_dirname, name, fname)
-        X, y = load_from_tsfile(abspath, return_data_type=return_type)
+        X, y = load_from_tsfile(abspath, return_data_type="nested_univ")
     # if split is None, load both train and test set
     elif split is None:
         fname = name + "_TRAIN.ts"
         abspath = os.path.join(local_module, local_dirname, name, fname)
-        X_train, y_train = load_from_tsfile(abspath, return_data_type=return_type)
+        X_train, y_train = load_from_tsfile(abspath, return_data_type="nested_univ")
 
         fname = name + "_TEST.ts"
         abspath = os.path.join(local_module, local_dirname, name, fname)
-        X_test, y_test = load_from_tsfile(abspath, return_data_type=return_type)
+        X_test, y_test = load_from_tsfile(abspath, return_data_type="nested_univ")
 
-        if isinstance(X_train, np.ndarray):
-            X = np.concatenate([X_train, X_test])
-        elif isinstance(X_train, pd.DataFrame):
-            X = pd.concat([X_train, X_test])
-            X = X.reset_index(drop=True)
-        else:
-            raise IOError(
-                f"Invalid data structure type {type(X_train)} for loading "
-                f"classification problem "
-            )
+        X = pd.concat([X_train, X_test])
+        X = X.reset_index(drop=True)
         y = np.concatenate([y_train, y_test])
 
     else:
         raise ValueError("Invalid `split` value =", split)
-    # Return appropriately
+
+    return_type = _alias_mtype_check(return_type)
     if return_X_y:
+        X = convert(X, from_type="nested_univ", to_type=return_type)
         return X, y
     else:
         X["class_val"] = pd.Series(y)
+        X = convert(X, from_type="nested_univ", to_type=return_type)
         return X
 
 
@@ -324,7 +330,7 @@ def load_from_tsfile(
             "numpy3D"/"numpy3d"/"np3D": 3D np.ndarray (instance, variable, time index)
             "numpy2d"/"np2d"/"numpyflat": 2D np.ndarray (instance, time index)
             "pd-multiindex": pd.DataFrame with 2-level (instance, time) MultiIndex
-        Exception is raised if the data cannot be stored in therequested type.
+        Exception is raised if the data cannot be stored in the requested type.
 
     Returns
     -------
@@ -343,12 +349,7 @@ def load_from_tsfile(
     ValueError if return_data_type = numpy2d but the data are multivariate and/
     or unequal length series
     """
-    if return_data_type is None:
-        return_data_type = "nested_univ"
-    if return_data_type in ["numpy2d", "numpy2D", "np2d", "np2D"]:
-        return_data_type = "numpyflat"
-    if return_data_type in ["numpy3d", "np3d", "np3D"]:
-        return_data_type = "numpy3D"
+    return_data_type = _alias_mtype_check(return_data_type)
 
     if not isinstance(return_data_type, str):
         raise TypeError(
diff --git a/sktime/datasets/_single_problem_loaders.py b/sktime/datasets/_single_problem_loaders.py
index 4f391d526..e1d014cd3 100644
--- a/sktime/datasets/_single_problem_loaders.py
+++ b/sktime/datasets/_single_problem_loaders.py
@@ -53,7 +53,9 @@ DIRNAME = "data"
 MODULE = os.path.dirname(__file__)
 
 
-def load_UCR_UEA_dataset(name, split=None, return_X_y=True, extract_path=None):
+def load_UCR_UEA_dataset(
+    name, split=None, return_X_y=True, return_type=None, extract_path=None
+):
     """Load dataset from UCR UEA time series archive.
 
     Downloads and extracts dataset if not already downloaded. Data is assumed to be
@@ -76,6 +78,18 @@ def load_UCR_UEA_dataset(name, split=None, return_X_y=True, extract_path=None):
         format <name>_TRAIN.ts or <name>_TEST.ts.
     return_X_y : bool, optional (default=False)
         it returns two objects, if False, it appends the class labels to the dataframe.
+    return_type: None or str, optional (default=None)
+        Memory data format specification to return X in, if None then return the default
+        "nested_univ" type.
+        str can be any other supported Panel mtype,
+            for list of mtypes, see datatypes.SCITYPE_REGISTER
+            for specifications, see examples/AA_datatypes_and_datasets.ipynb
+        commonly used specifications:
+            "nested_univ: nested pd.DataFrame, pd.Series in cells
+            "numpy3D"/"numpy3d"/"np3D": 3D np.ndarray (instance, variable, time index)
+            "numpy2d"/"np2d"/"numpyflat": 2D np.ndarray (instance, time index)
+            "pd-multiindex": pd.DataFrame with 2-level (instance, time) MultiIndex
+        Exception is raised if the data cannot be stored in the requested type.
     extract_path : str, optional (default=None)
         the path to look for the data. If no path is provided, the function
         looks in `sktime/datasets/data/`.
@@ -96,7 +110,7 @@ def load_UCR_UEA_dataset(name, split=None, return_X_y=True, extract_path=None):
     >>> from sktime.datasets import load_UCR_UEA_dataset
     >>> X, y = load_UCR_UEA_dataset(name="ArrowHead")
     """
-    return _load_dataset(name, split, return_X_y, extract_path)
+    return _load_dataset(name, split, return_X_y, return_type, extract_path)
 
 
 def load_plaid(split=None, return_X_y=True):
diff --git a/sktime/dists_kernels/__init__.py b/sktime/dists_kernels/__init__.py
index 81d54bdc3..18ff4b12c 100644
--- a/sktime/dists_kernels/__init__.py
+++ b/sktime/dists_kernels/__init__.py
@@ -5,7 +5,7 @@ from sktime.dists_kernels._base import (
     BasePairwiseTransformer,
     BasePairwiseTransformerPanel,
 )
-from sktime.dists_kernels.compose_tab_to_panel import AggrDist
+from sktime.dists_kernels.compose_tab_to_panel import AggrDist, FlatDist
 from sktime.dists_kernels.edit_dist import EditDist
 from sktime.dists_kernels.scipy_dist import ScipyDist
 
@@ -14,5 +14,6 @@ __all__ = [
     "BasePairwiseTransformerPanel",
     "AggrDist",
     "EditDist",
+    "FlatDist",
     "ScipyDist",
 ]
diff --git a/sktime/dists_kernels/_base.py b/sktime/dists_kernels/_base.py
index 8a2805496..3087f4c05 100644
--- a/sktime/dists_kernels/_base.py
+++ b/sktime/dists_kernels/_base.py
@@ -52,6 +52,8 @@ class BasePairwiseTransformer(BaseEstimator):
         "symmetric": False,  # is the transformer symmetric, i.e., t(x,y)=t(y,x) always?
         "X_inner_mtype": "numpy2D",  # which mtype is used internally in _transform?
         "fit_is_empty": True,  # is "fit" empty? Yes, for all pairwise transforms
+        "capability:missing_values": True,  # can estimator handle missing data?
+        "capability:multivariate": True,  # can estimator handle multivariate data?
     }
 
     def __init__(self):
@@ -183,6 +185,9 @@ class BasePairwiseTransformerPanel(BaseEstimator):
         "symmetric": False,  # is the transformer symmetric, i.e., t(x,y)=t(y,x) always?
         "X_inner_mtype": "df-list",  # which mtype is used internally in _transform?
         "fit_is_empty": True,  # is "fit" empty? Yes, for all pairwise transforms
+        "capability:missing_values": True,  # can estimator handle missing data?
+        "capability:multivariate": True,  # can estimator handle multivariate data?
+        "capability:unequal_length": False,  # can dist handle unequal length panels?
     }
 
     def __init__(self):
diff --git a/sktime/dists_kernels/compose.py b/sktime/dists_kernels/compose.py
new file mode 100644
index 000000000..9e613f3d2
--- /dev/null
+++ b/sktime/dists_kernels/compose.py
@@ -0,0 +1,220 @@
+# -*- coding: utf-8 -*-
+"""Pipelines for pairwise panel transformers."""
+
+__author__ = ["fkiraly"]
+
+from sktime.base import _HeterogenousMetaEstimator
+from sktime.dists_kernels._base import BasePairwiseTransformerPanel
+from sktime.transformations.base import BaseTransformer
+from sktime.transformations.compose import TransformerPipeline
+
+SUPPORTED_MTYPES = ["pd-multiindex", "nested_univ", "df-list", "numpy3D"]
+
+
+class PwTrafoPanelPipeline(BasePairwiseTransformerPanel, _HeterogenousMetaEstimator):
+    """Pipeline of transformers and a pairwise panel transformer.
+
+    `PwTrafoPanelPipeline` chains transformers and a pairwise transformer at the end.
+    The pipeline is constructed with a list of sktime transformers (BaseTransformer),
+        plus a pairwise panel transformer, following BasePairwiseTransformerPanel.
+    The transformer list can be unnamed - a simple list of transformers -
+        or string named - a list of pairs of string, estimator.
+
+    For a list of transformers `trafo1`, `trafo2`, ..., `trafoN` and an estimator `est`,
+        the pipeline behaves as follows:
+    `transform(X)` - running `trafo1.fit_transform` on `X`,
+        them `trafo2.fit_transform` on the output of `trafo1.fit_transform`, etc
+        sequentially, with `trafo[i]` receiving the output of `trafo[i-1]`.
+        Then passes output of `trafo[N]` to `pw_trafo.transform`, as `X`.
+        Same chain of transformers is run on `X2` and passed, if not `None`.
+
+    `PwTrafoPanelPipeline` can also be created by using the magic multiplication
+        on any parameter estimator: if `pw_t` is `BasePairwiseTransformerPanel`,
+            and `my_trafo1`, `my_trafo2` inherit from `BaseTransformer`, then,
+            for instance, `my_trafo1 * my_trafo2 * pw_t`
+            will result in the same object as  obtained from the constructor
+            `PwTrafoPanelPipeline(pw_trafo=pw_t, transformers=[my_trafo1, my_trafo2])`
+        magic multiplication can also be used with (str, transformer) pairs,
+            as long as one element in the chain is a transformer
+
+    Parameters
+    ----------
+    pw_trafo : pairwise panel transformer,
+        i.e., estimator inheriting from BasePairwiseTransformerPane
+        this is a "blueprint" estimator, state does not change when `fit` is called
+    transformers : list of sktime transformers, or
+        list of tuples (str, transformer) of sktime transformers
+        these are "blueprint" transformers, states do not change when `fit` is called
+
+    Examples
+    --------
+    >>> from sktime.dists_kernels.compose import PwTrafoPanelPipeline
+    >>> from sktime.dists_kernels.dtw import DtwDist
+    >>> from sktime.transformations.series.exponent import ExponentTransformer
+    >>> from sktime.datasets import load_unit_test
+    >>>
+    >>> X, _ = load_unit_test()
+    >>> X = X[0:3]
+    >>> pipeline = PwTrafoPanelPipeline(DtwDist(), [ExponentTransformer()])
+    >>> dist_mat = pipeline.transform(X)
+    """
+
+    _tags = {
+        "X_inner_mtype": SUPPORTED_MTYPES,
+        "capability:missing_values": True,  # can estimator handle missing data?
+        "capability:multivariate": True,  # can estimator handle multivariate data?
+        "capability:unequal_length": True,  # can dist handle unequal length panels?
+    }
+
+    def __init__(self, pw_trafo, transformers):
+
+        self.pw_trafo = pw_trafo
+        self.transformers = transformers
+        self.transformers_ = TransformerPipeline(transformers)
+
+        super(PwTrafoPanelPipeline, self).__init__()
+
+        # can handle multivariate iff: both classifier and all transformers can
+        multivariate = pw_trafo.get_tag("capability:multivariate", False)
+        multivariate = multivariate and not self.transformers_.get_tag(
+            "univariate-only", True
+        )
+        # can handle missing values iff: both classifier and all transformers can,
+        #   *or* transformer chain removes missing data
+        missing = pw_trafo.get_tag("capability:missing_values", False)
+        missing = missing and self.transformers_.get_tag("handles-missing-data", False)
+        missing = missing or self.transformers_.get_tag(
+            "capability:missing_values:removes", False
+        )
+
+        # set the pipeline tags to the above
+        tags_to_set = {
+            "capability:multivariate": multivariate,
+            "capability:missing_values": missing,
+        }
+        self.set_tags(**tags_to_set)
+
+    @property
+    def _transformers(self):
+        return self.transformers_._steps
+
+    @_transformers.setter
+    def _transformers(self, value):
+        self.transformers_._steps = value
+
+    def __rmul__(self, other):
+        """Magic * method, return concatenated PwTrafoPanelPipeline, trafos on left.
+
+        Implemented for `other` being a transformer, otherwise returns `NotImplemented`.
+
+        Parameters
+        ----------
+        other: `sktime` transformer, must inherit from BaseTransformer
+            otherwise, `NotImplemented` is returned
+
+        Returns
+        -------
+        ParamFitterPipeline object, concatenation of `other` (first) with `self` (last).
+        """
+        if isinstance(other, BaseTransformer):
+            # use the transformers dunder to get a TransformerPipeline
+            trafo_pipeline = other * self.transformers_
+            # then stick the expanded pipeline in a PwTrafoPanelPipeline
+            new_pipeline = PwTrafoPanelPipeline(
+                pw_trafo=self.pw_trafo,
+                transformers=trafo_pipeline.steps,
+            )
+            return new_pipeline
+        else:
+            return NotImplemented
+
+    def _transform(self, X, X2=None):
+        """Compute distance/kernel matrix.
+
+        private _transform containing core logic, called from public transform
+
+        Behaviour: returns pairwise distance/kernel matrix
+            between samples in X and X2
+                if X2 is not passed, is equal to X
+                if X/X2 is a pd.DataFrame and contains non-numeric columns,
+                    these are removed before computation
+
+        Parameters
+        ----------
+        X: sktime Panel data container
+        X2: sktime Panel data container
+
+        Returns
+        -------
+        distmat: np.array of shape [n, m]
+            (i,j)-th entry contains distance/kernel between X.iloc[i] and X2.iloc[j]
+        """
+        trafos = self.transformers_.clone()
+        pw_trafo = self.pw_trafo
+
+        Xt = trafos.fit_transform(X)
+
+        # find out whether we know that the resulting matrix is symmetric
+        #   since aligner distances are always symmetric,
+        #   we know it's the case for sure if X equals X2
+        if X2 is None:
+            X2t = None
+        else:
+            X2t = trafos.fit_transform(X2)
+
+        distmat = pw_trafo.transform(Xt, X2t)
+        return distmat
+
+    def get_params(self, deep=True):
+        """Get parameters of estimator in `transformers`.
+
+        Parameters
+        ----------
+        deep : boolean, optional, default=True
+            If True, will return the parameters for this estimator and
+            contained sub-objects that are estimators.
+
+        Returns
+        -------
+        params : mapping of string to any
+            Parameter names mapped to their values.
+        """
+        params = dict()
+        trafo_params = self._get_params("_transformers", deep=deep)
+        params.update(trafo_params)
+
+        return params
+
+    def set_params(self, **kwargs):
+        """Set the parameters of estimator in `transformers`.
+
+        Valid parameter keys can be listed with ``get_params()``.
+
+        Returns
+        -------
+        self : returns an instance of self.
+        """
+        if "pw_trafo" in kwargs.keys():
+            if not isinstance(kwargs["pw_trafo"], BasePairwiseTransformerPanel):
+                raise TypeError('"pw_trafo" arg must be a pairwise panel transformer')
+        trafo_keys = self._get_params("_transformers", deep=True).keys()
+        pw_keys = self.pw_trafo.get_params(deep=True).keys()
+        trafo_args = self._subset_dict_keys(dict_to_subset=kwargs, keys=trafo_keys)
+        pw_args = self._subset_dict_keys(dict_to_subset=kwargs, keys=pw_keys)
+        if len(pw_args) > 0:
+            self.pw_trafo.set_params(**pw_args)
+        if len(trafo_args) > 0:
+            self._set_params("_transformers", **trafo_args)
+        return self
+
+    @classmethod
+    def get_test_params(cls, parameter_set="default"):
+        """Test parameters for DistFromAligner."""
+        from sktime.dists_kernels.compose_tab_to_panel import AggrDist
+        from sktime.transformations.series.exponent import ExponentTransformer
+
+        params = {
+            "pw_trafo": AggrDist.create_test_instance(),
+            "transformers": [ExponentTransformer()],
+        }
+        return params
diff --git a/sktime/dists_kernels/compose_tab_to_panel.py b/sktime/dists_kernels/compose_tab_to_panel.py
index 046a2220d..51511b36b 100644
--- a/sktime/dists_kernels/compose_tab_to_panel.py
+++ b/sktime/dists_kernels/compose_tab_to_panel.py
@@ -5,6 +5,7 @@ Composers that create panel pairwise transformers from table pairwise transforme
 Currently implemented composers in this module:
 
     AggrDist - panel distance from aggregation of tabular distance matrix entries
+    FlatDist - panel distance from applying tabular distance to flattened panel matrix
 """
 
 __author__ = ["fkiraly"]
@@ -16,26 +17,38 @@ from sktime.utils._testing.deep_equals import deep_equals
 
 
 class AggrDist(BasePairwiseTransformerPanel):
-    """Panel distance from tabular distance aggregation.
+    r"""Panel distance from tabular distance aggregation.
 
     panel distance obtained by applying aggregation function to tabular distance matrix
         example: AggrDist(ScipyDist()) is mean Euclidean distance between series
 
-    Components
+    Formal details (for real valued objects, mixed typed rows in analogy):
+    Let :math:`d: \mathbb{R}^k \times \mathbb{R}^{k}\rightarrow \mathbb{R}`
+    be the pairwise function in `transformer`, when applied to `k`-vectors.
+    Let :math:`f:\mathbb{R}^{n \ times m}` be the function `aggfunc` when applied to
+    an :math:`(n \times m)` matrix.
+    Let :math:`x_1, \dots, x_N\in \mathbb{R}^{n \times k}`,
+    :math:`y_1, \dots y_M \in \mathbb{R}^{m \times k}` be collections of matrices,
+    representing time series panel valued inputs `X` and `X2`, as follows:
+    :math:`x_i` is the `i`-th instance in `X`, and :math:`x_{i, j\ell}` is the
+    `j`-th time point, `\ell`-th variable of `X`. Analogous for :math:`y` and `X2`.
+
+    Then, `transform(X, X2)` returns the :math:`(N \times M)` matrix
+    with :math:`(i, j)`-th entry :math:`f \left((d(x_{i, a}, y_{j, b}))_{a, b}\right)`,
+    where :math:`x_{i, a}` denotes the :math:`a`-th row of :math:`x_i`, and
+    :math:`y_{j, b}` denotes the :math:`b`-th row of :math:`x_j`.
+
+    Parameters
     ----------
     transformer: pairwise transformer of BasePairwiseTransformer scitype
-
-    Hyper-parameters
-    ----------------
-    aggfunc: aggregation function 2D np.array -> float
-        default = None, however, if transform is called then defaults to np.mean
-    aggfunc_is_symm: bool - whether aggregation function is symmetric
-                i.e., invariant under transposing argument, it always holds that
-                    aggfunc(matrix) = aggfunc(np.transpose(matrix))
+    aggfunc: aggregation function (2D np.array) -> float or None, optional
+        default = None = np.mean
+    aggfunc_is_symm: bool, optional, default=False
+        whether aggregation function is symmetric (should be set according to aggfunc)
+            i.e., invariant under transposing argument, it always holds that
+                aggfunc(matrix) = aggfunc(np.transpose(matrix))
             used for fast computation of the resultant matrix (if symmetric)
             if unknown, False is the "safe" option that ensures correctness
-        default = True if aggfunc default is used
-            False otherwise (should be set according to choice of aggfunc)
     """
 
     def __init__(
@@ -51,6 +64,9 @@ class AggrDist(BasePairwiseTransformerPanel):
 
         super(AggrDist, self).__init__()
 
+        if self.aggfunc_is_symm:
+            self.set_tag("symmetric", True)
+
     def _transform(self, X, X2=None):
         """Compute distance/kernel matrix.
 
@@ -110,3 +126,81 @@ class AggrDist(BasePairwiseTransformerPanel):
         from sktime.dists_kernels import ScipyDist
 
         return {"transformer": ScipyDist()}
+
+
+class FlatDist(BasePairwiseTransformerPanel):
+    r"""Panel distance from applying tabular distance to flattened time series.
+
+    Applies the wrapped tabular distance to flattened series.
+    Flattening is done to a 2D numpy array of shape (n_instances, (n_vars, n_timepts))
+
+    Formal details (for real valued objects, mixed typed rows in analogy):
+    Let :math:`d:\mathbb{R}^k \times \mathbb{R}^{k}\rightarrow \mathbb{R}`
+    be the pairwise function in `transformer`, when applied to `k`-vectors.
+    Let :math:`x_1, \dots, x_N\in \mathbb{R}^{n \times \ell}`,
+    :math:`y_1, \dots y_M \in \mathbb{R}^{n \times \ell}` be collections of matrices,
+    representing time series panel valued inputs `X` and `X2`, as follows:
+    :math:`x_i` is the `i`-th instance in `X`, and :math:`x_{i, j\ell}` is the
+    `j`-th time point, `\ell`-th variable of `X`. Analogous for :math:`y` and `X2`.
+    Let :math:`f:\mathbb{R}^{n \times \ell} \rightarrow \mathbb{R}^{n \cdot \ell}`
+    be the mapping that flattens matrices by column-first lexicographical ordering,
+    and assume :math:`k = n \cdot \ell`.
+
+    Then, `transform(X, X2)` returns the :math:`(N \times M)` matrix
+    with :math:`(i, j)`-th entry :math:`d\left(f(x_i), f(y_j)\right)`.
+
+    Parameters
+    ----------
+    transformer: pairwise transformer of BasePairwiseTransformer scitype
+    """
+
+    _tags = {
+        "X_inner_mtype": "numpy3D",  # which mtype is used internally in _transform?
+    }
+
+    def __init__(self, transformer):
+
+        self.transformer = transformer
+
+        super(FlatDist, self).__init__()
+
+    def _transform(self, X, X2=None):
+        """Compute distance/kernel matrix.
+
+            Core logic.
+
+        Behaviour: returns pairwise distance/kernel matrix
+            between samples in X and X2
+                if X2 is not passed, is equal to X
+                if X/X2 is a pd.DataFrame and contains non-numeric columns,
+                    these are removed before computation
+
+        Parameters
+        ----------
+        X: pd.DataFrame of length n, or 2D np.array with n rows
+        X2: pd.DataFrame of length m, or 2D np.array with m rows, optional
+            default X2 = X
+
+        Returns
+        -------
+        distmat: np.array of shape [n, m]
+            (i,j)-th entry contains distance/kernel between X.iloc[i] and X2.iloc[j]
+        """
+        n_inst, n_vars, n_tts = X.shape
+        X = X.reshape(n_inst, n_vars * n_tts)
+
+        n_inst2, n_vars2, n_tts2 = X2.shape
+        X2 = X2.reshape(n_inst2, n_vars2 * n_tts2)
+
+        if deep_equals(X, X2):
+            return self.transformer.transform(X)
+        else:
+            return self.transformer.transform(X, X2)
+
+    @classmethod
+    def get_test_params(cls, parameter_set="default"):
+        """Test parameters for FlatDist."""
+        # importing inside to avoid circular dependencies
+        from sktime.dists_kernels import ScipyDist
+
+        return {"transformer": ScipyDist()}
diff --git a/sktime/param_est/compose.py b/sktime/param_est/compose.py
index af717fabc..ccb14faff 100644
--- a/sktime/param_est/compose.py
+++ b/sktime/param_est/compose.py
@@ -25,11 +25,11 @@ class ParamFitterPipeline(BaseParamFitter, _HeterogenousMetaEstimator):
 
     For a list of transformers `trafo1`, `trafo2`, ..., `trafoN` and an estimator `est`,
         the pipeline behaves as follows:
-    `fit(X)` - changes styte by running `trafo1.fit_transform` on `X`,
+    `fit(X)` - changes state by running `trafo1.fit_transform` on `X`,
         them `trafo2.fit_transform` on the output of `trafo1.fit_transform`, etc
         sequentially, with `trafo[i]` receiving the output of `trafo[i-1]`,
         and then running `est.fit` with `X` being the output of `trafo[N]`
-    `update(X)` - changes styte by running `trafo1.update.transform` on `X`,
+    `update(X)` - changes state by running `trafo1.update.transform` on `X`,
         them `trafo2.update.transform` on the output of `trafo1.update.transform`, etc
         sequentially, with `trafo[i]` receiving the output of `trafo[i-1]`,
         and then running `est.update` with `X` being the output of `trafo[N]`
@@ -45,7 +45,7 @@ class ParamFitterPipeline(BaseParamFitter, _HeterogenousMetaEstimator):
             and `my_trafo1`, `my_trafo2` inherit from `BaseTransformer`, then,
             for instance, `my_trafo1 * my_trafo2 * est`
             will result in the same object as  obtained from the constructor
-            `ParamFitterPipeline(classifier=est, transformers=[my_trafo1, my_trafo2])`
+            `ParamFitterPipeline(param_est=est, transformers=[my_trafo1, my_trafo2])`
         magic multiplication can also be used with (str, transformer) pairs,
             as long as one element in the chain is a transformer
 
@@ -105,12 +105,12 @@ class ParamFitterPipeline(BaseParamFitter, _HeterogenousMetaEstimator):
 
         super(ParamFitterPipeline, self).__init__()
 
-        # can handle multivariate iff: both classifier and all transformers can
+        # can handle multivariate iff: both estimator and all transformers can
         multivariate = param_est.get_tag("capability:multivariate", False)
         multivariate = multivariate and not self.transformers_.get_tag(
             "univariate-only", True
         )
-        # can handle missing values iff: both classifier and all transformers can,
+        # can handle missing values iff: both estimator and all transformers can,
         #   *or* transformer chain removes missing data
         missing = param_est.get_tag("capability:missing_values", False)
         missing = missing and self.transformers_.get_tag("handles-missing-data", False)
@@ -150,9 +150,9 @@ class ParamFitterPipeline(BaseParamFitter, _HeterogenousMetaEstimator):
         if isinstance(other, BaseTransformer):
             # use the transformers dunder to get a TransformerPipeline
             trafo_pipeline = other * self.transformers_
-            # then stick the expanded pipeline in a ClassifierPipeline
+            # then stick the expanded pipeline in a ParamFitterPipeline
             new_pipeline = ParamFitterPipeline(
-                classifier=self.classifier,
+                param_est=self.param_est,
                 transformers=trafo_pipeline.steps,
             )
             return new_pipeline
@@ -250,7 +250,7 @@ class ParamFitterPipeline(BaseParamFitter, _HeterogenousMetaEstimator):
         -------
         self : returns an instance of self.
         """
-        if "classifier" in kwargs.keys():
+        if "param_est" in kwargs.keys():
             if not isinstance(kwargs["param_est"], BaseParamFitter):
                 raise TypeError('"param_est" arg must be an sktime parameter fitter')
         trafo_keys = self._get_params("_transformers", deep=True).keys()
@@ -272,10 +272,6 @@ class ParamFitterPipeline(BaseParamFitter, _HeterogenousMetaEstimator):
         parameter_set : str, default="default"
             Name of the set of test parameters to return, for use in tests. If no
             special parameters are defined for a value, will return `"default"` set.
-            For classifiers, a "default" set of parameters should be provided for
-            general testing, and a "results_comparison" set for comparing against
-            previously recorded results if the general set does not produce suitable
-            probabilities to compare against.
 
         Returns
         -------
diff --git a/sktime/registry/_tags.py b/sktime/registry/_tags.py
index 53b3f8db6..f5726beb6 100644
--- a/sktime/registry/_tags.py
+++ b/sktime/registry/_tags.py
@@ -201,13 +201,19 @@ ESTIMATOR_TAG_REGISTER = [
     ),
     (
         "capability:multivariate",
-        ["classifier", "early_classifier", "param_est"],
+        [
+            "classifier",
+            "early_classifier",
+            "param_est",
+            "transformer-pairwise",
+            "transformer-pairwise-panel",
+        ],
         "bool",
         "can the classifier classify time series with 2 or more variables?",
     ),
     (
         "capability:unequal_length",
-        ["classifier", "early_classifier", "transformer"],
+        ["classifier", "early_classifier", "transformer", "transformer-pairwise-panel"],
         "bool",
         "can the estimator handle unequal length time series?",
     ),
@@ -216,7 +222,13 @@ ESTIMATOR_TAG_REGISTER = [
     # Will be merged after refactor completion.
     (
         "capability:missing_values",
-        ["classifier", "early_classifier", "param_est"],
+        [
+            "classifier",
+            "early_classifier",
+            "param_est",
+            "transformer-pairwise",
+            "transformer-pairwise-panel",
+        ],
         "bool",
         "can the classifier handle missing data (NA, np.nan) in inputs?",
     ),


Test patch:
diff --git a/sktime/datasets/tests/test_data_io.py b/sktime/datasets/tests/test_data_io.py
index 584d99f1e..7e74c0546 100644
--- a/sktime/datasets/tests/test_data_io.py
+++ b/sktime/datasets/tests/test_data_io.py
@@ -1,12 +1,7 @@
 # -*- coding: utf-8 -*-
 """Test functions for data input and output."""
 
-__author__ = [
-    "SebasKoel",
-    "Emiliathewolf",
-    "TonyBagnall",
-    "jasonlines",
-]
+__author__ = ["SebasKoel", "Emiliathewolf", "TonyBagnall", "jasonlines", "achieveordie"]
 
 __all__ = []
 
@@ -30,8 +25,33 @@ from sktime.datasets import (
     load_uschange,
     write_dataframe_to_tsfile,
 )
-from sktime.datasets._data_io import MODULE, _convert_tsf_to_hierarchical
-from sktime.datatypes import check_is_mtype
+from sktime.datasets._data_io import (
+    MODULE,
+    _convert_tsf_to_hierarchical,
+    _load_provided_dataset,
+)
+from sktime.datatypes import MTYPE_LIST_PANEL, check_is_mtype
+
+# Disabling test for these mtypes since they don't support certain functionality yet
+_TO_DISABLE = ["pd-long", "pd-wide", "numpyflat"]
+
+
+@pytest.mark.parametrize("return_X_y", [True, False])
+@pytest.mark.parametrize(
+    "return_type", [mtype for mtype in MTYPE_LIST_PANEL if mtype not in _TO_DISABLE]
+)
+def test_load_provided_dataset(return_X_y, return_type):
+    """Test function to check for proper loading.
+
+    Check this via permutating between all possibilities of return_X_y and return_type.
+    """
+    if return_X_y:
+        X, y = _load_provided_dataset("UnitTest", "TRAIN", return_X_y, return_type)
+    else:
+        X = _load_provided_dataset("UnitTest", "TRAIN", return_X_y, return_type)
+
+    # Check whether object is same mtype or not, via bool
+    assert check_is_mtype(X, return_type)
 
 
 def test_load_from_tsfile():