Repository: mlflow/mlflow

Issue:
[BUG] log_params fails when logging a parameter that already exists, but without changing its value
### MLflow version

1.29.0

### System information

- **Python version**: 3.8


### Describe the problem

log_params fails when I try to log again a parameter (but without changing its initial value)

### Code to reproduce issue

```python
import mlflow
from mlflow.tracking.fluent import start_run
with start_run() as active_run:
    mlflow.log_params({"a": "1", "b": "2"})
    mlflow.log_params({"a": "1", "c": "2"})
```

You can reproduce this issue by adding a new test in tests/tracking/test_tracking.py with the code above.

### Stack trace

```python
  File "mlflow/mlflow/tracking/fluent.py", line 701, in log_params
    MlflowClient().log_batch(run_id=run_id, metrics=[], params=params_arr, tags=[])
  File "mlflow/mlflow/tracking/client.py", line 1034, in log_batch
    self._tracking_client.log_batch(run_id, metrics, params, tags)
  File "mlflow/mlflow/tracking/_tracking_service/client.py", line 372, in log_batch
    self.store.log_batch(
  File "mlflow/mlflow/store/tracking/sqlalchemy_store.py", line 1186, in log_batch
    raise e
  File "mlflow/mlflow/store/tracking/sqlalchemy_store.py", line 1182, in log_batch
    self._log_params(run_id, params)
  File "mlflow/mlflow/store/tracking/sqlalchemy_store.py", line 984, in _log_params
    raise MlflowException(
mlflow.exceptions.MlflowException: Changing param values is not allowed. Params were already logged='[{'key': 'c', 'old_value': None, 'new_value': '2'}]' for run ID='40d656aa104a4f7ca00bc9bf9aa99630'.
```

The error is misleading because 'c' was not logged before.

Discussion:
mlflow-automation: 
<!-- documentation preview -->

### Documentation preview will be available [here](https://output.circle-artifacts.com/output/job/88c9602b-0047-4692-971a-d202f4ef686d/artifacts/0/docs/build/html/index.html).

<details>
<summary>Notes</summary>

- Ignore this comment if this PR does not change the documentation.
- It takes a few minutes for the preview to be available.
- The preview is updated on every commit to this PR.
- Job URL: https://circleci.com/gh/mlflow/mlflow/33089
- Updated at: 2022-10-14 02:15:53.748603

</details>


Patch:
diff --git a/.gitignore b/.gitignore
index 5c57c80d2..abc7e0d76 100644
--- a/.gitignore
+++ b/.gitignore
@@ -80,6 +80,7 @@ venv.bak/
 *.swo
 *.idea
 *.vscode
+*.devcontainer
 *.iml
 *~
 
diff --git a/examples/pipelines/sklearn_regression_example b/examples/pipelines/sklearn_regression_example
index 196326364..c702852a7 160000
--- a/examples/pipelines/sklearn_regression_example
+++ b/examples/pipelines/sklearn_regression_example
@@ -1,1 +1,1 @@
-Subproject commit 19632636419a8471c1a3bbc7ee5e43e1bcba45e1
+Subproject commit c702852a7194439f488b8d9c5d9dbb9dabd84211
diff --git a/mlflow/pipelines/regression/v1/dag_help_strings.py b/mlflow/pipelines/regression/v1/dag_help_strings.py
index bbb553279..dc5de6379 100644
--- a/mlflow/pipelines/regression/v1/dag_help_strings.py
+++ b/mlflow/pipelines/regression/v1/dag_help_strings.py
@@ -45,15 +45,15 @@ metrics:
 """
 )
 
-INGEST_STEP = format_help_string(
-    """The 'ingest' step resolves the dataset specified by the 'data' section in pipeline.yaml and converts it to parquet format, leveraging the custom dataset parsing code defined in `steps/ingest.py` (and referred to by the 'custom_loader_method' attribute of the 'data' section in pipeline.yaml) if necessary. Subsequent steps convert this dataset into training, validation, & test sets and use them to develop a model. An example pipeline.yaml 'data' configuration is shown below.
+INGEST_STEP_BASE = """The '{0}' step resolves the dataset specified by the '{1}' section in pipeline.yaml and converts it to parquet format, leveraging the custom dataset parsing code defined in `steps/ingest.py` (and referred to by the 'custom_loader_method' attribute of the '{1}' section in pipeline.yaml) if necessary. Subsequent steps convert this dataset into training, validation, & test sets and use them to develop a model. An example pipeline.yaml '{1}' configuration is shown below.
 
-data:
+{1}:
   location: https://nyc-tlc.s3.amazonaws.com/trip+data/yellow_tripdata_2022-01.parquet
-  format: {{INGEST_DATA_FORMAT|default('parquet')}}
+  format: {{{{INGEST_DATA_FORMAT|default('parquet')}}}}
   custom_loader_method: steps.ingest.load_file_as_dataframe
 """
-)
+
+INGEST_STEP = format_help_string(INGEST_STEP_BASE.format("ingest", "data"))
 
 INGEST_USER_CODE = format_help_string(
     """\"\"\"\nsteps/ingest.py defines customizable logic for parsing arbitrary dataset formats (i.e. formats that are not natively parsed by MLflow Pipelines) via the `load_file_as_dataframe` function. Note that the Parquet, Delta, and Spark SQL dataset formats are natively parsed by MLflow Pipelines, and you do not need to define custom logic for parsing them. An example `load_file_as_dataframe` implementation is displayed below (note that a different function name or module can be specified via the 'custom_loader_method' attribute of the 'data' section in pipeline.yaml).\n\"\"\"\n
@@ -245,16 +245,26 @@ register:
 """
 )
 
+INGEST_SCORING_STEP = format_help_string(INGEST_STEP_BASE.format("ingest_scoring", "data_scoring"))
+
+INGESTED_SCORING_DATA = format_help_string(
+    "The ingested parquet representation of the dataset defined in the 'data_scoring' section of pipeline.yaml. Subsequent steps scores this dataset for batch scoring."
+)
+
 PREDICT_STEP = format_help_string(
-    """The 'predict' step applies the specified model against the cleaned dataset produced by the 'ingest_scoring' step. An example pipeline.yaml 'predict' step definition is shown below.
+    """The 'predict' uses the model produced by the 'register' step to score the ingested dataset produced by the 'ingest_scoring' step and writes the resulting dataset to the specified output format and location. Namely, it uses the latest version of the registered model specified by the `model_name` attribute of the pipeline.yaml 'register' step definition. To fix a specific model for use in the 'predict' step, provide its model URI as the 'model_uri' attribute of the pipeline.yaml 'predict' step definition. An example pipeline.yaml 'predict' step definition is shown below.
 steps:
   predict:
-    model_uri: "models:/taxi_fare_regressor/Production"
+    model_uri: "models:/taxi_fare_regressor/Production" # optional
     output_format: {{OUTPUT_DATA_FORMAT|default('parquet')}}
     output_location: "{{OUTPUT_DATA_LOCATION}}"
 """
 )
 
+SCORED_DATA = format_help_string(
+    "The dataset produced by scoring the ingested dataset generated by the 'ingest_scoring' step with the model specified by the the 'predict' step."
+)
+
 REGISTERED_MODEL_VERSION = format_help_string(
     "The Model Version in the MLflow Model Registry corresponding to the trained model. A Model Version is produced if the trained model meets the defined performance criteria for model registration or if `allow_non_validated_model: true` is specified in the 'register' step definition of pipeline.yaml"
 )
diff --git a/mlflow/pipelines/regression/v1/pipeline.py b/mlflow/pipelines/regression/v1/pipeline.py
index a80c54794..b13fc0bb4 100644
--- a/mlflow/pipelines/regression/v1/pipeline.py
+++ b/mlflow/pipelines/regression/v1/pipeline.py
@@ -331,10 +331,22 @@ class RegressionPipeline(_BasePipeline):
                     "help_string": dag_help_strings.REGISTERED_MODEL_VERSION,
                     "help_string_type": "text",
                 },
+                "ingest_scoring_step_help": {
+                    "help_string": dag_help_strings.INGEST_SCORING_STEP,
+                    "help_string_type": "text",
+                },
+                "ingested_scoring_data_help": {
+                    "help_string": dag_help_strings.INGESTED_SCORING_DATA,
+                    "help_string_type": "text",
+                },
                 "predict_step_help": {
                     "help_string": dag_help_strings.PREDICT_STEP,
                     "help_string_type": "text",
                 },
+                "scored_data_help": {
+                    "help_string": dag_help_strings.SCORED_DATA,
+                    "help_string_type": "text",
+                },
             }
         )
 
diff --git a/mlflow/pipelines/regression/v1/resources/pipeline_dag_template.html b/mlflow/pipelines/regression/v1/resources/pipeline_dag_template.html
index c0d6d07ec..3c928ef80 100644
--- a/mlflow/pipelines/regression/v1/resources/pipeline_dag_template.html
+++ b/mlflow/pipelines/regression/v1/resources/pipeline_dag_template.html
@@ -48,11 +48,20 @@
 
     <div class="wrapper">
       <div class="mermaid pane-loading">
-          flowchart TB
-          pipeline([pipeline.yaml])
+          flowchart TD
 
-          ingestUserCode([steps/ingest.py]) --> ingestMLPStep[["<font>ingest</font>"]]
-          ingestMLPStep --> dataParquet[(ingested_data)]
+          subgraph " "
+            pipeline([pipeline.yaml])
+
+            ingestUserCode([steps/ingest.py]) --> ingestMLPStep[["<font>ingest</font>"]]
+            ingestMLPStep --> dataParquet[(ingested_data)]
+
+            ingestScoringUserCode([steps/ingest.py]) --> ingestScoringMLPStep[["<font>ingest_scoring</font>"]]
+            ingestScoringMLPStep --> dataScoringParquet[(ingested_scoring_data)]
+
+            dataScoring[(ingested_scoring_data)] --> predictMLPStep[["<font>predict</font>"]]
+            predictMLPStep --> dataScored[(scored_data)]
+          end
 
           data[(ingested_data)] --> splitStep[["<font>split</font>"]]
           transformUserCode([steps/transform.py]) --> transformMLPStep
@@ -95,6 +104,9 @@
           click registerMLPStep renderMoreInformation "{{register_step_help}}"
           click customMetricsUserCode renderMoreInformation "{{custom_metrics_user_code_help}}"
           click splitUserCode renderMoreInformation "{{split_user_code_help}}"
+          click ingestScoringUserCode renderMoreInformation "{{ingest_user_code_help}}"
+          click ingestScoringMLPStep renderMoreInformation "{{ingest_scoring_step_help}}"
+          click predictMLPStep renderMoreInformation "{{predict_step_help}}"
 
           click pipeline renderMoreInformation "{{pipeline_yaml_help}}"
           click dataParquet renderMoreInformation "{{ingested_data_help}}"
@@ -108,6 +120,9 @@
           click transformer renderMoreInformation "{{fitted_transformer_help}}"
           click model_validation_status renderMoreInformation "{{model_validation_status_help}}"
           click registered_model_version renderMoreInformation "{{registered_model_version_help}}"
+          click dataScoringParquet renderMoreInformation "{{ingested_scoring_data_help}}"
+          click dataScoring renderMoreInformation "{{ingested_scoring_data_help}}"
+          click dataScored renderMoreInformation "{{scored_data_help}}"
 
       </div>
       <div id="editor"></div>
@@ -130,6 +145,7 @@
             startOnLoad:true,
             securityLevel:'loose',
             theme: (window.matchMedia && window.matchMedia("(prefers-color-scheme: dark)").matches) ? "dark" : "default",
+            themeCSS: ".cluster rect { fill: none; stroke: none; }",
             flowchart:{
                 useMaxWidth:true,
                 htmlLabels:true,
diff --git a/mlflow/pipelines/steps/automl/__init__.py b/mlflow/pipelines/steps/automl/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/mlflow/pipelines/steps/automl/flaml.py b/mlflow/pipelines/steps/automl/flaml.py
new file mode 100644
index 000000000..07d8af236
--- /dev/null
+++ b/mlflow/pipelines/steps/automl/flaml.py
@@ -0,0 +1,123 @@
+import logging
+from typing import Dict, Any, Tuple
+
+import pandas as pd
+from sklearn.base import BaseEstimator
+
+import mlflow
+from mlflow import MlflowException
+from mlflow.models.evaluation.default_evaluator import _get_regressor_metrics
+from mlflow.pipelines.utils.metrics import PipelineMetric, _load_custom_metric_functions
+
+_logger = logging.getLogger(__name__)
+
+_AUTOML_DEFAULT_TIME_BUDGET = 10
+_MLFLOW_TO_FLAML_METRICS = {
+    "mean_absolute_error": "mae",
+    "mean_squared_error": "mse",
+    "root_mean_squared_error": "rmse",
+    "r2_score": "r2",
+    "mean_absolute_percentage_error": "mape",
+}
+
+
+def get_estimator_and_best_params(
+    X,
+    y,
+    task: str,
+    step_config: Dict[str, Any],
+    pipeline_root: str,
+    evaluation_metrics: Dict[str, PipelineMetric],
+    primary_metric: str,
+) -> Tuple[BaseEstimator, Dict[str, Any]]:
+    return _create_model_automl(
+        X, y, task, step_config, pipeline_root, evaluation_metrics, primary_metric
+    )
+
+
+def _create_custom_metric_flaml(metric_name: str, coeff: int, custom_func: callable) -> callable:
+    def add_suffix(metrics: Dict[str, float], suffix: str) -> Dict[str, float]:
+        return {f"{k}_{suffix}": v for k, v in metrics.items()}
+
+    def calc_metric(X, y, estimator) -> Dict[str, float]:
+        y_pred = estimator.predict(X)
+        builtin_metrics = _get_regressor_metrics(y, y_pred, sample_weights=None)
+        res_df = pd.DataFrame()
+        res_df["prediction"] = y_pred
+        res_df["target"] = y.values
+        return custom_func(res_df, builtin_metrics)
+
+    # pylint: disable=keyword-arg-before-vararg
+    # pylint: disable=unused-argument
+    def custom_metric(
+        X_val,
+        y_val,
+        estimator,
+        labels,
+        X_train,
+        y_train,
+        weight_val=None,
+        weight_train=None,
+        *args,
+    ):
+        val_metrics = calc_metric(X_val, y_val, estimator)
+        if metric_name not in val_metrics:
+            raise MlflowException(
+                f"User-defined function has not calculated expected primary metric {metric_name}.\n"
+                f"The function has returned the following metrics: {val_metrics}"
+            )
+        main_metric = coeff * val_metrics[metric_name]
+        val_metrics = add_suffix(val_metrics, "val")
+        train_metrics = add_suffix(calc_metric(X_train, y_train, estimator), "train")
+        return main_metric, {**val_metrics, **train_metrics}
+
+    return custom_metric
+
+
+def _create_model_automl(
+    X,
+    y,
+    task: str,
+    step_config: Dict[str, Any],
+    pipeline_root: str,
+    evaluation_metrics: Dict[str, PipelineMetric],
+    primary_metric: str,
+) -> Tuple[BaseEstimator, Dict[str, Any]]:
+    try:
+        from flaml import AutoML
+    except ImportError:
+        raise MlflowException("Please install FLAML to use AutoML!")
+
+    try:
+        if primary_metric in _MLFLOW_TO_FLAML_METRICS and primary_metric in evaluation_metrics:
+            metric = _MLFLOW_TO_FLAML_METRICS[primary_metric]
+        elif primary_metric in evaluation_metrics:
+            metric = _create_custom_metric_flaml(
+                primary_metric,
+                -1 if evaluation_metrics[primary_metric].greater_is_better else 1,
+                _load_custom_metric_functions(pipeline_root, [evaluation_metrics[primary_metric]])[
+                    0
+                ],
+            )
+        else:
+            raise MlflowException(
+                f"There is no FLAML alternative or custom metric for {primary_metric} metric."
+            )
+
+        automl_settings = step_config.get("flaml_params", {})
+        automl_settings["time_budget"] = step_config.get(
+            "time_budget_secs", _AUTOML_DEFAULT_TIME_BUDGET
+        )
+        automl_settings["metric"] = metric
+        automl_settings["task"] = task
+        # Disabled Autologging, because during the hyperparameter search
+        # it tries to log the same parameters multiple times.
+        mlflow.autolog(disable=True)
+        automl = AutoML()
+        automl.fit(X, y, **automl_settings)
+        mlflow.autolog(disable=False, log_models=False)
+        return automl.model.estimator, automl.best_config
+    except Exception as e:
+        raise MlflowException(
+            f"Error has occurred during training of AutoML model using FLAML: {repr(e)}"
+        )
diff --git a/mlflow/pipelines/steps/train.py b/mlflow/pipelines/steps/train.py
index b06985c04..b3b05ec57 100644
--- a/mlflow/pipelines/steps/train.py
+++ b/mlflow/pipelines/steps/train.py
@@ -52,7 +52,6 @@ _logger = logging.getLogger(__name__)
 
 
 class TrainStep(BaseStep):
-
     MODEL_ARTIFACT_RELATIVE_PATH = "model"
 
     def __init__(self, step_config, pipeline_root, pipeline_config=None):
@@ -61,8 +60,9 @@ class TrainStep(BaseStep):
         self.pipeline_config = pipeline_config
 
     def _validate_and_apply_step_config(self):
+        self.task = self.step_config.get("template_name", "regression/v1").rsplit("/", 1)[0]
         if "using" in self.step_config:
-            if self.step_config["using"] not in ["estimator_spec"]:
+            if self.step_config["using"] not in ["estimator_spec", "automl/flaml"]:
                 raise MlflowException(
                     f"Invalid train step configuration value {self.step_config['using']} for "
                     f"key 'using'. Supported values are: ['estimator_spec']",
@@ -135,14 +135,16 @@ class TrainStep(BaseStep):
                 error_code=INVALID_PARAMETER_VALUE,
             )
         self.skip_data_profiling = self.step_config.get("skip_data_profiling", False)
-        if "estimator_method" not in self.step_config:
+        if (
+            "estimator_method" not in self.step_config
+            and self.step_config["using"] == "estimator_spec"
+        ):
             raise MlflowException(
-                "Missing 'estimator_method' configuration in the train step.",
+                "Missing 'estimator_method' configuration in the train step, "
+                "which is using 'estimator_spec'.",
                 error_code=INVALID_PARAMETER_VALUE,
             )
-        self.train_module_name, self.estimator_method_name = self.step_config[
-            "estimator_method"
-        ].rsplit(".", 1)
+
         self.primary_metric = _get_primary_metric(self.step_config)
         self.user_defined_custom_metrics = {
             metric.name: metric for metric in _get_custom_metrics(self.step_config)
@@ -247,11 +249,6 @@ class TrainStep(BaseStep):
             relative_path="transformer.pkl",
         )
 
-        sys.path.append(self.pipeline_root)
-        estimator_fn = getattr(
-            importlib.import_module(self.train_module_name), self.estimator_method_name
-        )
-
         tags = {
             MLFLOW_SOURCE_TYPE: SourceType.to_string(SourceType.PIPELINE),
             MLFLOW_PIPELINE_TEMPLATE_NAME: self.step_config["template_name"],
@@ -264,29 +261,13 @@ class TrainStep(BaseStep):
         best_estimator_params = None
         mlflow.autolog(log_models=False, silent=True)
         with mlflow.start_run(tags=tags) as run:
-            estimator_hardcoded_params = self.step_config["estimator_params"]
-            if self.step_config["tuning_enabled"]:
-                best_estimator_params = self._tune_and_get_best_estimator_params(
-                    run.info.run_id,
-                    estimator_hardcoded_params,
-                    estimator_fn,
-                    X_train,
-                    y_train,
-                    validation_df,
-                    output_directory,
-                )
-                estimator = estimator_fn(best_estimator_params)
-            elif len(estimator_hardcoded_params) > 0:
-                self._write_best_parameters_outputs(
-                    {}, estimator_hardcoded_params, output_directory
-                )
-                estimator = estimator_fn(estimator_hardcoded_params)
-            else:
-                estimator = estimator_fn()
 
+            estimator = self._resolve_estimator(
+                X_train, y_train, validation_df, run, output_directory
+            )
             estimator.fit(X_train, y_train)
 
-            logged_estimator = self._log_estimator_to_mlflow(estimator, X_train, tags)
+            logged_estimator = self._log_estimator_to_mlflow(estimator, X_train)
 
             # Create a pipeline consisting of the transformer+model for test data evaluation
             with open(transformer_path, "rb") as f:
@@ -387,6 +368,65 @@ class TrainStep(BaseStep):
 
         return card
 
+    def _get_user_defined_estimator(self, X_train, y_train, validation_df, run, output_directory):
+        train_module_name, estimator_method_name = self.step_config["estimator_method"].rsplit(
+            ".", 1
+        )
+        sys.path.append(self.pipeline_root)
+        estimator_fn = getattr(importlib.import_module(train_module_name), estimator_method_name)
+        estimator_hardcoded_params = self.step_config["estimator_params"]
+        if self.step_config["tuning_enabled"]:
+            best_estimator_params = self._tune_and_get_best_estimator_params(
+                run.info.run_id,
+                estimator_hardcoded_params,
+                estimator_fn,
+                X_train,
+                y_train,
+                validation_df,
+                output_directory,
+            )
+            estimator = estimator_fn(best_estimator_params)
+        elif len(estimator_hardcoded_params) > 0:
+            self._write_best_parameters_outputs(
+                {}, estimator_hardcoded_params, {}, output_directory
+            )
+            estimator = estimator_fn(estimator_hardcoded_params)
+        else:
+            estimator = estimator_fn()
+        return estimator
+
+    def _resolve_estimator_plugin(self, plugin_str, X_train, y_train, output_directory):
+        plugin_str = plugin_str.replace("/", ".")
+        estimator_fn = getattr(
+            importlib.import_module(f"mlflow.pipelines.steps.{plugin_str}"),
+            "get_estimator_and_best_params",
+        )
+        estimator, best_parameters = estimator_fn(
+            X_train,
+            y_train,
+            self.task,
+            self.step_config,
+            self.pipeline_root,
+            self.evaluation_metrics,
+            self.primary_metric,
+        )
+        self.best_estimator_name = estimator.__class__.__name__
+        self.best_estimator_class = (
+            f"{estimator.__class__.__module__}.{estimator.__class__.__name__}"
+        )
+        self.best_parameters = best_parameters
+        self._write_best_parameters_outputs({}, {}, best_parameters, output_directory)
+        return estimator
+
+    def _resolve_estimator(self, X_train, y_train, validation_df, run, output_directory):
+        using_plugin = self.step_config.get("using", "estimator_spec")
+        if using_plugin == "estimator_spec":
+            return self._get_user_defined_estimator(
+                X_train, y_train, validation_df, run, output_directory
+            )
+        else:
+            return self._resolve_estimator_plugin(using_plugin, X_train, y_train, output_directory)
+
     def _get_leaderboard_df(self, run, eval_metrics):
         import pandas as pd
 
@@ -629,17 +669,31 @@ class TrainStep(BaseStep):
                 )
             )
 
-        # Tab 8: Best Parameters
+        # Tab 8: Best Parameters (AutoML and Tuning)
+        is_automl_run = self.step_config["using"].startswith("automl")
         best_parameters_yaml = os.path.join(output_directory, "best_parameters.yaml")
+
         if os.path.exists(best_parameters_yaml):
             best_parameters_card_tab = card.add_tab(
-                "Best Parameters",
+                f"Best Parameters {' (AutoML)' if is_automl_run else ''}",
                 "{{ BEST_PARAMETERS }} ",
             )
+
+            if is_automl_run:
+                automl_estimator_str = (
+                    f"<b>Best estimator:</b><br>"
+                    f"<pre>{self.best_estimator_name}</pre><br>"
+                    f"<b>Best estimator class:</b><br>"
+                    f"<pre>{self.best_estimator_class}</pre><br><br>"
+                )
+            else:
+                automl_estimator_str = ""
+
             best_parameters = open(best_parameters_yaml).read()
             best_parameters_card_tab.add_html(
                 "BEST_PARAMETERS",
-                f"<b>Best parameters:</b><br>" f"<pre>{best_parameters}</pre><br><br>",
+                f"{automl_estimator_str}<b>Best parameters:</b><br>"
+                f"<pre>{best_parameters}</pre><br><br>",
             )
 
         # Tab 9: HP trials
@@ -881,13 +935,15 @@ class TrainStep(BaseStep):
         else:
             best_hardcoded_params = {}
         best_combined_params = dict(estimator_hardcoded_params, **best_hp_params)
-        self._write_best_parameters_outputs(best_hp_params, best_hardcoded_params, output_directory)
+        self._write_best_parameters_outputs(
+            best_hp_params, best_hardcoded_params, {}, output_directory
+        )
         return best_combined_params
 
     def _log_estimator_to_mlflow(self, estimator, X_train_sampled, on_worker=False):
         from mlflow.models.signature import infer_signature
 
-        if hasattr(estimator, "best_score_"):
+        if hasattr(estimator, "best_score_") and (type(estimator.best_score_) in [int, float]):
             mlflow.log_metric("best_cv_score", estimator.best_score_)
         if hasattr(estimator, "best_params_"):
             mlflow.log_params(estimator.best_params_)
@@ -913,19 +969,22 @@ class TrainStep(BaseStep):
         return logged_estimator
 
     def _write_best_parameters_outputs(
-        self, best_hp_params, best_hardcoded_params, output_directory
+        self, best_hp_params, best_hardcoded_params, automl_params, output_directory
     ):
-        best_parameters_path = os.path.join(output_directory, "best_parameters.yaml")
-        if os.path.exists(best_parameters_path):
-            os.remove(best_parameters_path)
-        with open(best_parameters_path, "a") as file:
-            file.write("# tuned hyperparameters\n")
-            self._safe_dump_with_numeric_values(best_hp_params, file, default_flow_style=False)
-            file.write("\n# hardcoded parameters\n")
-            self._safe_dump_with_numeric_values(
-                best_hardcoded_params, file, default_flow_style=False
-            )
-        mlflow.log_artifact(best_parameters_path, artifact_path="train")
+        if best_hp_params or best_hardcoded_params or automl_params:
+            best_parameters_path = os.path.join(output_directory, "best_parameters.yaml")
+            if os.path.exists(best_parameters_path):
+                os.remove(best_parameters_path)
+            with open(best_parameters_path, "a") as file:
+                self._write_one_param_output(automl_params, file, "automl parameters")
+                self._write_one_param_output(best_hp_params, file, "tuned hyperparameters")
+                self._write_one_param_output(best_hardcoded_params, file, "hardcoded parameters")
+            mlflow.log_artifact(best_parameters_path, artifact_path="train")
+
+    def _write_one_param_output(self, params, file, caption):
+        if params:
+            file.write(f"# {caption} \n")
+            self._safe_dump_with_numeric_values(params, file, default_flow_style=False)
 
     def _safe_dump_with_numeric_values(self, data, file, **kwargs):
         import numpy as np
diff --git a/mlflow/server/handlers.py b/mlflow/server/handlers.py
index 816252f11..04dc11238 100644
--- a/mlflow/server/handlers.py
+++ b/mlflow/server/handlers.py
@@ -675,9 +675,7 @@ def _create_run():
     )
 
     tags = [RunTag(tag.key, tag.value) for tag in request_message.tags]
-    run_name = (
-        _generate_random_name() if request_message.run_name is None else request_message.run_name
-    )
+    run_name = _generate_random_name() if not request_message.run_name else request_message.run_name
     run = _get_tracking_store().create_run(
         experiment_id=request_message.experiment_id,
         user_id=request_message.user_id,
diff --git a/mlflow/store/tracking/file_store.py b/mlflow/store/tracking/file_store.py
index 06667849e..14dbe515e 100644
--- a/mlflow/store/tracking/file_store.py
+++ b/mlflow/store/tracking/file_store.py
@@ -582,7 +582,7 @@ class FileStore(AbstractStore):
         run_info = self._get_run_info(run_id)
         check_run_is_active(run_info)
         new_info = run_info._copy_with_overrides(run_status, end_time, run_name=run_name)
-        if run_name is not None:
+        if run_name:
             self._set_run_tag(run_info, RunTag(MLFLOW_RUN_NAME, run_name))
         self._overwrite_run_info(new_info)
         return new_info
@@ -604,7 +604,7 @@ class FileStore(AbstractStore):
                 "Could not create run under non-active experiment with ID %s." % experiment_id,
                 databricks_pb2.INVALID_STATE,
             )
-        run_name = run_name if run_name is not None else _generate_random_name()
+        run_name = run_name if run_name else _generate_random_name()
         run_uuid = uuid.uuid4().hex
         artifact_uri = self._get_artifact_dir(experiment_id, run_uuid)
         run_info = RunInfo(
diff --git a/mlflow/store/tracking/sqlalchemy_store.py b/mlflow/store/tracking/sqlalchemy_store.py
index e8dae6c29..65ec65a79 100644
--- a/mlflow/store/tracking/sqlalchemy_store.py
+++ b/mlflow/store/tracking/sqlalchemy_store.py
@@ -545,7 +545,7 @@ class SqlAlchemyStore(AbstractStore):
             artifact_location = append_to_uri_path(
                 experiment.artifact_location, run_id, SqlAlchemyStore.ARTIFACTS_FOLDER_NAME
             )
-            run_name = run_name if run_name is not None else _generate_random_name()
+            run_name = run_name if run_name else _generate_random_name()
             run = SqlRun(
                 name=run_name,
                 artifact_uri=artifact_location,
@@ -633,7 +633,7 @@ class SqlAlchemyStore(AbstractStore):
             self._check_run_is_active(run)
             run.status = RunStatus.to_string(run_status)
             run.end_time = end_time
-            if run_name is not None:
+            if run_name:
                 run.name = run_name
                 run_name_tag = self._try_get_run_tag(session, run_id, MLFLOW_RUN_NAME)
                 if run_name_tag is None:
@@ -962,52 +962,36 @@ class SqlAlchemyStore(AbstractStore):
         if not params:
             return
 
-        param_instances = [
-            SqlParam(run_uuid=run_id, key=param.key, value=param.value) for param in params
-        ]
-
         with self.ManagedSessionMaker() as session:
             run = self._get_run(run_uuid=run_id, session=session)
             self._check_run_is_active(run)
-            # commit the session to make sure that we catch any IntegrityError
-            # and try to handle them.
-            try:
-                self._save_to_db(session=session, objs=param_instances)
-                session.commit()
-            except sqlalchemy.exc.IntegrityError:
-                # Roll back the current session to make it usable for further transactions. In the
-                # event of an error during "commit", a rollback is required in order to continue
-                # using the session. In this case, we re-use the session because the SqlRun, `run`,
-                # is lazily evaluated during the invocation of `run.params`.
-                session.rollback()
-
-                # in case of an integrity error, compare the parameters of the
-                # run. If the parameters match the ones whom being saved,
-                # ignore the exception since idempotency is reached.
-                # Also, multiple params for the same key can still be passed within
-                # the same batch. So, handle them by selecting the first param
-                # for the given key
-                run_params = {param.key: param.value for param in run.params}
-                non_matching_params = []
-                for param in param_instances:
-                    existing_value = run_params.get(param.key)
-                    if param.value != existing_value:
+            existing_params = {p.key: p.value for p in run.params}
+            new_params = []
+            non_matching_params = []
+            for param in params:
+                if param.key in existing_params:
+                    if param.value != existing_params[param.key]:
                         non_matching_params.append(
                             {
                                 "key": param.key,
-                                "old_value": existing_value,
+                                "old_value": existing_params[param.key],
                                 "new_value": param.value,
                             }
                         )
+                    continue
+                new_params.append(SqlParam(run_uuid=run_id, key=param.key, value=param.value))
 
-                if non_matching_params:
-                    raise MlflowException(
-                        "Changing param values is not allowed. Params were already logged='{}'"
-                        " for run ID='{}'.".format(non_matching_params, run_id),
-                        INVALID_PARAMETER_VALUE,
-                    )
-                # if there's no mismatch, do not raise an Exception since
-                # we are sure that idempotency is reached.
+            if non_matching_params:
+                raise MlflowException(
+                    "Changing param values is not allowed. Params were already logged='{}'"
+                    " for run ID='{}'.".format(non_matching_params, run_id),
+                    INVALID_PARAMETER_VALUE,
+                )
+
+            if not new_params:
+                return
+
+            self._save_to_db(session=session, objs=new_params)
 
     def set_experiment_tag(self, experiment_id, tag):
         """
diff --git a/mlflow/utils/databricks_utils.py b/mlflow/utils/databricks_utils.py
index cd3585f2c..f0201fecf 100644
--- a/mlflow/utils/databricks_utils.py
+++ b/mlflow/utils/databricks_utils.py
@@ -648,6 +648,6 @@ def _construct_databricks_model_version_url(
     if workspace_id and workspace_id != "0":
         model_version_url += "?o=" + str(workspace_id)
 
-    model_version_url += f"#mlflow/models/{name}/{version}"
+    model_version_url += f"#mlflow/models/{name}/versions/{version}"
 
     return model_version_url


Test patch:
diff --git a/requirements/test-requirements.txt b/requirements/test-requirements.txt
index e14d90d6a..c2b027383 100644
--- a/requirements/test-requirements.txt
+++ b/requirements/test-requirements.txt
@@ -22,4 +22,6 @@ werkzeug>=2.2.0
 # Required by mlflow.keras / mlflow.pytorch tests.
 transformers
 # Required by tuning tests
-hyperopt
\ No newline at end of file
+hyperopt
+# Required by automl tests
+flaml
\ No newline at end of file
diff --git a/tests/pipelines/test_register_step.py b/tests/pipelines/test_register_step.py
index 231aab000..00ea2542b 100644
--- a/tests/pipelines/test_register_step.py
+++ b/tests/pipelines/test_register_step.py
@@ -314,7 +314,7 @@ steps:
     with open(register_step_output_dir / "card.html", "r") as f:
         step_card_content = f.read()
 
-    assert f"<a href={workspace_url}#mlflow/models/demo_model/1" in step_card_content
+    assert f"<a href={workspace_url}#mlflow/models/demo_model/versions/1" in step_card_content
     assert (
         f"<a href={workspace_url}#mlflow/experiments/1/runs/{run_id}/artifactPath/train/model>"
         in step_card_content
diff --git a/tests/pipelines/test_train_step.py b/tests/pipelines/test_train_step.py
index 0886ed79c..14724e87a 100644
--- a/tests/pipelines/test_train_step.py
+++ b/tests/pipelines/test_train_step.py
@@ -60,7 +60,9 @@ def setup_train_dataset(pipeline_root: Path):
 
 
 # Sets up the constructed TrainStep instance
-def setup_train_step(pipeline_root: Path, use_tuning: bool, with_hardcoded_params: bool = True):
+def setup_train_step_with_tuning(
+    pipeline_root: Path, use_tuning: bool, with_hardcoded_params: bool = True
+):
     pipeline_yaml = pipeline_root.joinpath(_PIPELINE_CONFIG_FILE_NAME)
     if with_hardcoded_params:
         estimator_params = """
@@ -134,6 +136,48 @@ def setup_train_step(pipeline_root: Path, use_tuning: bool, with_hardcoded_param
     return train_step
 
 
+def setup_train_step_with_automl(
+    pipeline_root: Path,
+    primary_metric: str = "root_mean_squared_error",
+    generate_custom_metrics: bool = False,
+):
+    pipeline_yaml = pipeline_root.joinpath(_PIPELINE_CONFIG_FILE_NAME)
+
+    custom_metric = """
+    custom:
+      - name: weighted_mean_squared_error
+        function: weighted_mean_squared_error
+        greater_is_better: False
+            """
+    pipeline_yaml.write_text(
+        f"""
+  template: regression/v1
+  target_col: y
+  profile: test_profile
+  run_args:
+    step: train
+  experiment:
+    name: demo
+    tracking_uri: { mlflow.get_tracking_uri() }
+  steps:
+    train:
+      using: automl/flaml
+      time_budget_secs: 20
+      flaml_params:
+        estimator_list:
+          - xgboost
+          - rf
+          - lgbm
+  metrics:
+    { custom_metric if generate_custom_metrics else "" }
+    primary: { primary_metric }
+        """
+    )
+    pipeline_config = read_yaml(pipeline_root, _PIPELINE_CONFIG_FILE_NAME)
+    train_step = TrainStep.from_pipeline_config(pipeline_config, str(pipeline_root))
+    return train_step
+
+
 def estimator_fn(estimator_params=None):
     from sklearn.linear_model import SGDRegressor
 
@@ -150,7 +194,7 @@ def test_train_steps_writes_model_pkl_and_card(tmp_pipeline_root_path, use_tunin
         os.environ, {_MLFLOW_PIPELINES_EXECUTION_DIRECTORY_ENV_VAR: str(tmp_pipeline_root_path)}
     ):
         train_step_output_dir = setup_train_dataset(tmp_pipeline_root_path)
-        train_step = setup_train_step(tmp_pipeline_root_path, use_tuning)
+        train_step = setup_train_step_with_tuning(tmp_pipeline_root_path, use_tuning)
         train_step.run(str(train_step_output_dir))
 
     assert (train_step_output_dir / "model/model.pkl").exists()
@@ -170,7 +214,7 @@ def test_train_steps_writes_card_with_model_and_run_links_on_databricks(
     monkeypatch.setenv(_MLFLOW_PIPELINES_EXECUTION_DIRECTORY_ENV_VAR, str(tmp_pipeline_root_path))
 
     train_step_output_dir = setup_train_dataset(tmp_pipeline_root_path)
-    train_step = setup_train_step(tmp_pipeline_root_path, use_tuning)
+    train_step = setup_train_step_with_tuning(tmp_pipeline_root_path, use_tuning)
     train_step.run(str(train_step_output_dir))
 
     with open(train_step_output_dir / "run_id") as f:
@@ -193,7 +237,7 @@ def test_train_steps_autologs(tmp_pipeline_root_path, use_tuning):
         os.environ, {_MLFLOW_PIPELINES_EXECUTION_DIRECTORY_ENV_VAR: str(tmp_pipeline_root_path)}
     ):
         train_step_output_dir = setup_train_dataset(tmp_pipeline_root_path)
-        train_step = setup_train_step(tmp_pipeline_root_path, use_tuning)
+        train_step = setup_train_step_with_tuning(tmp_pipeline_root_path, use_tuning)
         train_step.run(str(train_step_output_dir))
 
     assert os.path.exists(train_step_output_dir / "run_id")
@@ -218,7 +262,7 @@ def test_train_steps_with_correct_tags(tmp_pipeline_root_path, use_tuning):
         },
     ):
         train_step_output_dir = setup_train_dataset(tmp_pipeline_root_path)
-        train_step = setup_train_step(tmp_pipeline_root_path, use_tuning)
+        train_step = setup_train_step_with_tuning(tmp_pipeline_root_path, use_tuning)
         train_step.run(str(train_step_output_dir))
 
     assert os.path.exists(train_step_output_dir / "run_id")
@@ -239,7 +283,7 @@ def test_train_step_with_tuning_best_parameters(tmp_pipeline_root_path):
         os.environ, {_MLFLOW_PIPELINES_EXECUTION_DIRECTORY_ENV_VAR: str(tmp_pipeline_root_path)}
     ):
         train_step_output_dir = setup_train_dataset(tmp_pipeline_root_path)
-        train_step = setup_train_step(tmp_pipeline_root_path, use_tuning=True)
+        train_step = setup_train_step_with_tuning(tmp_pipeline_root_path, use_tuning=True)
         train_step.run(str(train_step_output_dir))
     assert (train_step_output_dir / "best_parameters.yaml").exists()
 
@@ -266,7 +310,7 @@ def test_train_step_with_tuning_output_yaml_correct(
         os.environ, {_MLFLOW_PIPELINES_EXECUTION_DIRECTORY_ENV_VAR: str(tmp_pipeline_root_path)}
     ):
         train_step_output_dir = setup_train_dataset(tmp_pipeline_root_path)
-        train_step = setup_train_step(
+        train_step = setup_train_step_with_tuning(
             tmp_pipeline_root_path, use_tuning=True, with_hardcoded_params=with_hardcoded_params
         )
         train_step.run(str(train_step_output_dir))
@@ -274,9 +318,12 @@ def test_train_step_with_tuning_output_yaml_correct(
 
     with open(os.path.join(train_step_output_dir, "best_parameters.yaml")) as f:
         lines = f.readlines()
-        assert lines[0] == "# tuned hyperparameters\n"
-        assert lines[expected_num_tuned + 2] == "# hardcoded parameters\n"
-        assert len(lines) == expected_num_tuned + expected_num_hardcoded + 3
+        assert lines[0] == "# tuned hyperparameters \n"
+        if with_hardcoded_params:
+            assert lines[expected_num_tuned + 1] == "# hardcoded parameters \n"
+        assert len(lines) == expected_num_tuned + expected_num_hardcoded + (
+            2 if with_hardcoded_params else 1
+        )
 
 
 def test_train_step_with_tuning_child_runs_and_early_stop(tmp_pipeline_root_path):
@@ -284,7 +331,7 @@ def test_train_step_with_tuning_child_runs_and_early_stop(tmp_pipeline_root_path
         os.environ, {_MLFLOW_PIPELINES_EXECUTION_DIRECTORY_ENV_VAR: str(tmp_pipeline_root_path)}
     ):
         train_step_output_dir = setup_train_dataset(tmp_pipeline_root_path)
-        train_step = setup_train_step(tmp_pipeline_root_path, use_tuning=True)
+        train_step = setup_train_step_with_tuning(tmp_pipeline_root_path, use_tuning=True)
         train_step.run(str(train_step_output_dir))
 
     with open(train_step_output_dir / "run_id") as f:
@@ -321,3 +368,58 @@ def test_search_space(tmp_pipeline_root_path):
 @pytest.mark.parametrize("tuning_param,logged_param", [(1, "1"), (1.0, "1.0"), ("a", " a ")])
 def test_tuning_param_equal(tuning_param, logged_param):
     assert TrainStep.is_tuning_param_equal(tuning_param, logged_param)
+
+
+@pytest.mark.parametrize(
+    ("automl", "primary_metric", "generate_custom_metrics"),
+    [
+        (False, "root_mean_squared_error", False),
+        (True, "root_mean_squared_error", False),
+        (True, "weighted_mean_squared_error", True),
+    ],
+)
+def test_automl(tmp_pipeline_root_path, automl, primary_metric, generate_custom_metrics):
+    with mock.patch.dict(
+        os.environ,
+        {
+            _MLFLOW_PIPELINES_EXECUTION_DIRECTORY_ENV_VAR: str(tmp_pipeline_root_path),
+            _MLFLOW_PIPELINES_EXECUTION_TARGET_STEP_NAME_ENV_VAR: "train",
+        },
+    ):
+        train_step_output_dir = setup_train_dataset(tmp_pipeline_root_path)
+        if generate_custom_metrics:
+            pipeline_steps_dir = tmp_pipeline_root_path.joinpath("steps")
+            pipeline_steps_dir.joinpath("custom_metrics.py").write_text(
+                """
+def weighted_mean_squared_error(eval_df, builtin_metrics):
+    from sklearn.metrics import mean_squared_error
+
+    return {
+        "weighted_mean_squared_error": mean_squared_error(
+            eval_df["prediction"],
+            eval_df["target"],
+            sample_weight=1 / eval_df["prediction"].values,
+        )
+    }
+        """
+            )
+        if automl:
+            train_step = setup_train_step_with_automl(
+                tmp_pipeline_root_path,
+                primary_metric=primary_metric,
+                generate_custom_metrics=generate_custom_metrics,
+            )
+        else:
+            train_step = setup_train_step_with_tuning(
+                tmp_pipeline_root_path,
+                use_tuning=True,
+                with_hardcoded_params=False,
+            )
+        train_step._validate_and_apply_step_config()
+        train_step._run(str(train_step_output_dir))
+
+        with open(train_step_output_dir / "run_id") as f:
+            run_id = f.read()
+
+        metrics = MlflowClient().get_run(run_id).data.metrics
+        assert f"{primary_metric}_on_data_training" in metrics
diff --git a/tests/store/tracking/test_file_store.py b/tests/store/tracking/test_file_store.py
index 1b9e08422..ef18a4a6c 100644
--- a/tests/store/tracking/test_file_store.py
+++ b/tests/store/tracking/test_file_store.py
@@ -795,6 +795,16 @@ class TestFileStore(unittest.TestCase, AbstractStoreTest):
         assert isinstance(tags_run.data, RunData)
         assert tags_run.data.tags == {**tags_dict, MLFLOW_RUN_NAME: tags_run.info.run_name}
 
+        name_empty_str_run = fs.create_run(
+            experiment_id=FileStore.DEFAULT_EXPERIMENT_ID,
+            user_id="user",
+            start_time=0,
+            tags=tags_entities,
+            run_name="",
+        )
+        run_name = name_empty_str_run.info.run_name
+        assert run_name.split("-")[0] in _GENERATOR_PREDICATES
+
     def test_create_run_sets_name(self):
         fs = FileStore(self.test_root)
         run = fs.create_run(
@@ -905,6 +915,19 @@ class TestFileStore(unittest.TestCase, AbstractStoreTest):
         get_run = fs.get_run(run_id)
         assert get_run.info.run_name == "first name"
 
+    def test_update_run_does_not_rename_run_with_empty_string_name(self):
+        fs = FileStore(self.test_root)
+        run_id = fs.create_run(
+            experiment_id=FileStore.DEFAULT_EXPERIMENT_ID,
+            user_id="user",
+            start_time=0,
+            tags=[],
+            run_name="first name",
+        ).info.run_id
+        fs.update_run_info(run_id, RunStatus.FINISHED, 1000, "")
+        get_run = fs.get_run(run_id)
+        assert get_run.info.run_name == "first name"
+
     def test_list_run_infos(self):
         fs = FileStore(self.test_root)
         for exp_id in self.experiments:
diff --git a/tests/store/tracking/test_sqlalchemy_store.py b/tests/store/tracking/test_sqlalchemy_store.py
index 0ef0cf171..a92c3c24d 100644
--- a/tests/store/tracking/test_sqlalchemy_store.py
+++ b/tests/store/tracking/test_sqlalchemy_store.py
@@ -935,6 +935,10 @@ class TestSqlAlchemyStore(unittest.TestCase, AbstractStoreTest):
         run = self.store.get_run(run_id)
         assert run.info.run_name.split("-")[0] in _GENERATOR_PREDICATES
 
+        name_empty_str_run = self.store.create_run(**{**configs, **{"run_name": ""}})
+        run_name = name_empty_str_run.info.run_name
+        assert run_name.split("-")[0] in _GENERATOR_PREDICATES
+
     def test_to_mlflow_entity_and_proto(self):
         # Create a run and log metrics, params, tags to the run
         created_run = self._run_factory()
@@ -1381,6 +1385,11 @@ class TestSqlAlchemyStore(unittest.TestCase, AbstractStoreTest):
         self.assertEqual(run.info.run_name, "new name")
         self.assertEqual(run.data.tags.get(mlflow_tags.MLFLOW_RUN_NAME), "new name")
 
+        self.store.update_run_info(run_id, RunStatus.FINISHED, 1000, "")
+        run = self.store.get_run(run_id)
+        self.assertEqual(run.info.run_name, "new name")
+        self.assertEqual(run.data.tags.get(mlflow_tags.MLFLOW_RUN_NAME), "new name")
+
         self.store.delete_tag(run_id, mlflow_tags.MLFLOW_RUN_NAME)
         run = self.store.get_run(run_id)
         self.assertEqual(run.info.run_name, "new name")
@@ -2218,6 +2227,35 @@ class TestSqlAlchemyStore(unittest.TestCase, AbstractStoreTest):
         assert exception_context.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
         self._verify_logged(self.store, run.info.run_id, metrics=[], params=[param], tags=[])
 
+    def test_log_batch_with_unchanged_and_new_params(self):
+        """
+        Test case to ensure the following code works:
+        ---------------------------------------------
+        mlflow.log_params({"a": 0, "b": 1})
+        mlflow.log_params({"a": 0, "c": 2})
+        ---------------------------------------------
+        """
+        run = self._run_factory()
+        self.store.log_batch(
+            run.info.run_id,
+            metrics=[],
+            params=[entities.Param("a", "0"), entities.Param("b", "1")],
+            tags=[],
+        )
+        self.store.log_batch(
+            run.info.run_id,
+            metrics=[],
+            params=[entities.Param("a", "0"), entities.Param("c", "2")],
+            tags=[],
+        )
+        self._verify_logged(
+            self.store,
+            run.info.run_id,
+            metrics=[],
+            params=[entities.Param("a", "0"), entities.Param("b", "1"), entities.Param("c", "2")],
+            tags=[],
+        )
+
     def test_log_batch_param_overwrite_disallowed_single_req(self):
         # Test that attempting to overwrite a param via log_batch results in an exception
         run = self._run_factory()