diff --git a/src/preset_cli/cli/main.py b/src/preset_cli/cli/main.py
index 40829a7e..01bcb4c4 100644
--- a/src/preset_cli/cli/main.py
+++ b/src/preset_cli/cli/main.py
@@ -169,7 +169,7 @@ def preset_cli(  # pylint: disable=too-many-branches, too-many-locals, too-many-
             click.echo(
                 click.style(
                     "Failed to auth using the provided credentials."
-                    " Please run 'preset-cli auth'",
+                    " Please run ``preset-cli auth``",
                     fg="bright_red",
                 ),
             )
@@ -255,7 +255,7 @@ def auth(baseurl: str, overwrite: bool = False, show: bool = False) -> None:
             click.style(
                 (
                     f"The file {credentials_path} already exists. "
-                    "Pass --overwrite to replace it."
+                    "Pass ``--overwrite`` to replace it."
                 ),
                 fg="bright_red",
             ),
@@ -353,14 +353,13 @@ def list_group_membership(
     if save_report and save_report.casefold() not in {"yaml", "csv"}:
         click.echo(
             click.style(
-                "Invalid option. Please use --save-report=csv or --save-report=yaml",
+                "Invalid option. Please use ``--save-report=csv`` or ``--save-report=yaml``",
                 fg="bright_red",
             ),
         )
         sys.exit(1)
 
     for team in teams:
-
         # print the team name in case multiple teams were provided and it's not an export
         if not save_report and len(teams) > 1:
             click.echo(f"## Team {team} ##")
@@ -371,12 +370,10 @@ def list_group_membership(
 
         # account for pagination
         while start_at <= group_count:
-
             groups = client.get_group_membership(team, start_at)
             group_count = groups["totalResults"]
 
             if group_count > 0:
-
                 # print groups in console
                 if not save_report:
                     print_group_membership(groups)
@@ -430,10 +427,8 @@ def export_group_membership_csv(groups: Dict[str, Any], team: str) -> None:
     """
     csv_name = team + "_user_group_membership.csv"
     for group in groups["Resources"]:
-
         # CSV report would include a group only in case it has members
         if group.get("members"):
-
             # Assure we just write headers once
             file_exists = os.path.isfile(csv_name)
 
diff --git a/src/preset_cli/cli/superset/export.py b/src/preset_cli/cli/superset/export.py
index f2cb7175..c59ce0fe 100644
--- a/src/preset_cli/cli/superset/export.py
+++ b/src/preset_cli/cli/superset/export.py
@@ -135,7 +135,7 @@ def export_resource(  # pylint: disable=too-many-arguments, too-many-locals
         target = root / file_name
         if target.exists() and not overwrite:
             raise Exception(
-                f"File already exists and --overwrite was not specified: {target}",
+                f"File already exists and ``--overwrite`` was not specified: {target}",
             )
         if not target.parent.exists():
             target.parent.mkdir(parents=True, exist_ok=True)
diff --git a/src/preset_cli/cli/superset/sync/dbt/command.py b/src/preset_cli/cli/superset/sync/dbt/command.py
index 8546c4d2..a84c1b05 100644
--- a/src/preset_cli/cli/superset/sync/dbt/command.py
+++ b/src/preset_cli/cli/superset/sync/dbt/command.py
@@ -71,10 +71,22 @@
     "--preserve-columns",
     is_flag=True,
     default=False,
-    help="Preserve column configurations",
+    help="Preserve column and metric configurations defined in Preset",
+)
+@click.option(
+    "--preserve-metadata",
+    is_flag=True,
+    default=False,
+    help="Preserve column and metric configurations defined in Preset",
+)
+@click.option(
+    "--merge-metadata",
+    is_flag=True,
+    default=False,
+    help="Update Preset configurations based on dbt metadata. Preset-only metrics are preserved",
 )
 @click.pass_context
-def dbt_core(  # pylint: disable=too-many-arguments, too-many-locals
+def dbt_core(  # pylint: disable=too-many-arguments, too-many-branches, too-many-locals ,too-many-statements
     ctx: click.core.Context,
     file: str,
     project: Optional[str],
@@ -88,6 +100,8 @@ def dbt_core(  # pylint: disable=too-many-arguments, too-many-locals
     external_url_prefix: str = "",
     exposures_only: bool = False,
     preserve_columns: bool = False,
+    preserve_metadata: bool = False,
+    merge_metadata: bool = False,
 ) -> None:
     """
     Sync models/metrics from dbt Core to Superset and charts/dashboards to dbt exposures.
@@ -96,7 +110,20 @@ def dbt_core(  # pylint: disable=too-many-arguments, too-many-locals
     url = URL(ctx.obj["INSTANCE"])
     client = SupersetClient(url, auth)
 
-    reload_columns = not preserve_columns
+    if (preserve_columns or preserve_metadata) and merge_metadata:
+        click.echo(
+            click.style(
+                """
+                ``--preserve-columns`` / ``--preserve-metadata`` and ``--merge-metadata``
+                can't be combined. Please include only one to the command.
+                """,
+                fg="bright_red",
+            ),
+        )
+        sys.exit(1)
+
+    reload_columns = not (preserve_columns or preserve_metadata or merge_metadata)
+    preserve_metadata = preserve_columns if preserve_columns else preserve_metadata
 
     if profiles is None:
         profiles = os.path.expanduser("~/.dbt/profiles.yml")
@@ -134,7 +161,7 @@ def dbt_core(  # pylint: disable=too-many-arguments, too-many-locals
     else:
         click.echo(
             click.style(
-                "FILE should be either manifest.json or dbt_project.yml",
+                "FILE should be either ``manifest.json`` or ``dbt_project.yml``",
                 fg="bright_red",
             ),
         )
@@ -185,7 +212,7 @@ def dbt_core(  # pylint: disable=too-many-arguments, too-many-locals
                 external_url_prefix,
             )
         except DatabaseNotFoundError:
-            click.echo("No database was found, pass --import-db to create")
+            click.echo("No database was found, pass ``--import-db`` to create")
             return
 
         datasets = sync_datasets(
@@ -196,6 +223,7 @@ def dbt_core(  # pylint: disable=too-many-arguments, too-many-locals
             disallow_edits,
             external_url_prefix,
             reload_columns=reload_columns,
+            merge_metadata=merge_metadata,
         )
 
     if exposures:
@@ -325,7 +353,19 @@ def get_job_id(
     "--preserve-columns",
     is_flag=True,
     default=False,
-    help="Preserve column configurations",
+    help="Preserve column and metric configurations defined in Preset",
+)
+@click.option(
+    "--preserve-metadata",
+    is_flag=True,
+    default=False,
+    help="Preserve column and metric configurations defined in Preset",
+)
+@click.option(
+    "--merge-metadata",
+    is_flag=True,
+    default=False,
+    help="Update Preset configurations based on dbt metadata. Preset-only metrics are preserved",
 )
 @click.pass_context
 def dbt_cloud(  # pylint: disable=too-many-arguments, too-many-locals
@@ -339,6 +379,8 @@ def dbt_cloud(  # pylint: disable=too-many-arguments, too-many-locals
     external_url_prefix: str = "",
     exposures_only: bool = False,
     preserve_columns: bool = False,
+    preserve_metadata: bool = False,
+    merge_metadata: bool = False,
 ) -> None:
     """
     Sync models/metrics from dbt Cloud to Superset.
@@ -350,7 +392,20 @@ def dbt_cloud(  # pylint: disable=too-many-arguments, too-many-locals
     dbt_auth = TokenAuth(token)
     dbt_client = DBTClient(dbt_auth)
 
-    reload_columns = not preserve_columns
+    if (preserve_columns or preserve_metadata) and merge_metadata:
+        click.echo(
+            click.style(
+                """
+                ``--preserve-columns`` / ``--preserve-metadata`` and ``--merge-metadata``
+                can't be combined. Please include only one to the command.
+                """,
+                fg="bright_red",
+            ),
+        )
+        sys.exit(1)
+
+    reload_columns = not (preserve_columns or preserve_metadata or merge_metadata)
+    preserve_metadata = preserve_columns if preserve_columns else preserve_metadata
 
     if job_id is None:
         job_id = get_job_id(dbt_client)
@@ -390,6 +445,7 @@ def dbt_cloud(  # pylint: disable=too-many-arguments, too-many-locals
             disallow_edits,
             external_url_prefix,
             reload_columns=reload_columns,
+            merge_metadata=merge_metadata,
         )
 
     if exposures:
diff --git a/src/preset_cli/cli/superset/sync/dbt/datasets.py b/src/preset_cli/cli/superset/sync/dbt/datasets.py
index cc3befdf..2d028667 100644
--- a/src/preset_cli/cli/superset/sync/dbt/datasets.py
+++ b/src/preset_cli/cli/superset/sync/dbt/datasets.py
@@ -36,6 +36,18 @@ def model_in_database(model: ModelSchema, url: SQLAlchemyURL) -> bool:
     return model["database"] == url.database
 
 
+def clean_metadata(metadata: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Remove incompatbile columns from metatada.
+    When updating an existing column/metric we need to remove some fields from the payload.
+    """
+    for key in ("changed_on", "created_on", "type_generic"):
+        if key in metadata:
+            del metadata[key]
+
+    return metadata
+
+
 def create_dataset(
     client: SupersetClient,
     database: Dict[str, Any],
@@ -68,7 +80,7 @@ def create_dataset(
     return client.create_dataset(**kwargs)
 
 
-def sync_datasets(  # pylint: disable=too-many-locals, too-many-branches, too-many-arguments, too-many-statements
+def sync_datasets(  # pylint: disable=too-many-locals, too-many-branches, too-many-arguments, too-many-statements # noqa:C901
     client: SupersetClient,
     models: List[ModelSchema],
     metrics: List[MetricSchema],
@@ -77,6 +89,7 @@ def sync_datasets(  # pylint: disable=too-many-locals, too-many-branches, too-ma
     external_url_prefix: str,
     certification: Optional[Dict[str, Any]] = None,
     reload_columns: bool = True,
+    merge_metadata: bool = False,
 ) -> List[Any]:
     """
     Read the dbt manifest and import models as datasets with metrics.
@@ -86,7 +99,6 @@ def sync_datasets(  # pylint: disable=too-many-locals, too-many-branches, too-ma
     # add datasets
     datasets = []
     for model in models:
-
         # load additional metadata from dbt model definition
         model_kwargs = model.get("meta", {}).pop("superset", {})
 
@@ -132,14 +144,28 @@ def sync_datasets(  # pylint: disable=too-many-locals, too-many-branches, too-ma
         }
 
         dataset_metrics = []
+        current_metrics = {}
         model_metrics = {
             metric["name"]: metric for metric in get_metrics_for_model(model, metrics)
         }
+
+        if not reload_columns:
+            current_metrics = {
+                metric["metric_name"]: metric
+                for metric in client.get_dataset(dataset["id"])["metrics"]
+            }
+            for name, metric in current_metrics.items():
+                # remove data that is not part of the update payload
+                metric = clean_metadata(metric)
+                if not merge_metadata or name not in model_metrics:
+                    dataset_metrics.append(metric)
+
         for name, metric in model_metrics.items():
             meta = metric.get("meta", {})
             kwargs = meta.pop("superset", {})
-            dataset_metrics.append(
-                {
+
+            if reload_columns or name not in current_metrics or merge_metadata:
+                metric_definition = {
                     "expression": get_metric_expression(name, model_metrics),
                     "metric_name": name,
                     "metric_type": (
@@ -150,15 +176,17 @@ def sync_datasets(  # pylint: disable=too-many-locals, too-many-branches, too-ma
                     "description": metric.get("description", ""),
                     "extra": json.dumps(meta),
                     **kwargs,  # include additional metric metadata defined in metric.meta.superset
-                },
-            )
+                }
+                if merge_metadata and name in current_metrics:
+                    metric_definition["id"] = current_metrics[name]["id"]
+                dataset_metrics.append(metric_definition)
 
         # update dataset metadata from dbt and clearing metrics
         update = {
             "description": model.get("description", ""),
             "extra": json.dumps(extra),
             "is_managed_externally": disallow_edits,
-            "metrics": [],
+            "metrics": [] if reload_columns else dataset_metrics,
             **model_kwargs,  # include additional model metadata defined in model.meta.superset
         }
         if base_url:
@@ -166,8 +194,7 @@ def sync_datasets(  # pylint: disable=too-many-locals, too-many-branches, too-ma
             update["external_url"] = str(base_url.with_fragment(fragment))
         client.update_dataset(dataset["id"], override_columns=reload_columns, **update)
 
-        # ...then update metrics
-        if dataset_metrics:
+        if reload_columns and dataset_metrics:
             update = {
                 "metrics": dataset_metrics,
             }
@@ -183,10 +210,8 @@ def sync_datasets(  # pylint: disable=too-many-locals, too-many-branches, too-ma
                     column["description"] = column_metadata[name].get("description", "")
                     column["verbose_name"] = column_metadata[name].get("name", "")
 
-                # remove columns that are not part of the update payload
-                for key in ("changed_on", "created_on", "type_generic"):
-                    if key in column:
-                        del column[key]
+                # remove data that is not part of the update payload
+                column = clean_metadata(column)
 
                 # for some reason this is being sent as null sometimes
                 # https://github.com/preset-io/backend-sdk/issues/163
diff --git a/src/preset_cli/cli/superset/sync/native/command.py b/src/preset_cli/cli/superset/sync/native/command.py
index d27dc59a..20461d52 100644
--- a/src/preset_cli/cli/superset/sync/native/command.py
+++ b/src/preset_cli/cli/superset/sync/native/command.py
@@ -375,7 +375,7 @@ def import_resources(
         click.echo(
             click.style(
                 (
-                    "The following file(s) already exist. Pass --overwrite to "
+                    "The following file(s) already exist. Pass ``--overwrite`` to "
                     f"replace them.\n{existing_list}"
                 ),
                 fg="bright_red",
diff --git a/tests/cli/main_test.py b/tests/cli/main_test.py
index cfb96a14..230b6563 100644
--- a/tests/cli/main_test.py
+++ b/tests/cli/main_test.py
@@ -1204,7 +1204,7 @@ def test_list_group_membership_group_with_no_members(mocker: MockerFixture) -> N
 
 def test_list_group_membership_incorrect_export(mocker: MockerFixture) -> None:
     """
-    Test the ``list_group_membership`` command with an incorrect --export-report parameter.
+    Test the ``list_group_membership`` command with an incorrect ``--export-report`` parameter.
     """
     PresetClient = mocker.patch("preset_cli.cli.main.PresetClient")
     mocker.patch("preset_cli.cli.main.input", side_effect=["invalid", "-"])
@@ -1229,7 +1229,7 @@ def test_list_group_membership_incorrect_export(mocker: MockerFixture) -> None:
 
 def test_list_group_membership_export_yaml(mocker: MockerFixture) -> None:
     """
-    Test the ``list_group_membership`` command setting --export-report=yaml.
+    Test the ``list_group_membership`` command setting ``--export-report=yaml``.
     """
     PresetClient = mocker.patch("preset_cli.cli.main.PresetClient")
     mocker.patch("preset_cli.cli.main.input", side_effect=["invalid", "-"])
@@ -1315,7 +1315,7 @@ def test_list_group_membership_export_yaml(mocker: MockerFixture) -> None:
 
 def test_list_group_membership_export_csv(mocker: MockerFixture) -> None:
     """
-    Test the ``list_group_membership`` setting --export-report=csv.
+    Test the ``list_group_membership`` setting ``--export-report=csv``.
     """
     PresetClient = mocker.patch("preset_cli.cli.main.PresetClient")
     mocker.patch("preset_cli.cli.main.input", side_effect=["invalid", "-"])
diff --git a/tests/cli/superset/export_test.py b/tests/cli/superset/export_test.py
index 4ff2c1df..a1304149 100644
--- a/tests/cli/superset/export_test.py
+++ b/tests/cli/superset/export_test.py
@@ -144,7 +144,7 @@ def test_export_resource_overwrite(
             disable_jinja_escaping=False,
         )
     assert str(excinfo.value) == (
-        "File already exists and --overwrite was not specified: "
+        "File already exists and ``--overwrite`` was not specified: "
         "/path/to/root/databases/gsheets.yaml"
     )
 
@@ -533,7 +533,7 @@ def test_export_resource_jinja_escaping_disabled(
     dataset_export: BytesIO,
 ) -> None:
     """
-    Test ``export_resource`` with --disable-jinja-escaping.
+    Test ``export_resource`` with ``--disable-jinja-escaping``.
     """
     root = Path("/path/to/root")
     fs.create_dir(root)
@@ -574,7 +574,7 @@ def test_export_resource_jinja_escaping_disabled_command(
     fs: FakeFilesystem,
 ) -> None:
     """
-    Test the ``export_assets`` with --disable-jinja-escaping command.
+    Test the ``export_assets`` with ``--disable-jinja-escaping`` command.
     """
     # root must exist for command to succeed
     root = Path("/path/to/root")
diff --git a/tests/cli/superset/sync/dbt/command_test.py b/tests/cli/superset/sync/dbt/command_test.py
index 4eefde2d..650c1b35 100644
--- a/tests/cli/superset/sync/dbt/command_test.py
+++ b/tests/cli/superset/sync/dbt/command_test.py
@@ -215,6 +215,7 @@ def test_dbt_core(mocker: MockerFixture, fs: FakeFilesystem) -> None:
         False,
         "",
         reload_columns=True,
+        merge_metadata=False,
     )
     sync_exposures.assert_called_with(
         client,
@@ -224,12 +225,12 @@ def test_dbt_core(mocker: MockerFixture, fs: FakeFilesystem) -> None:
     )
 
 
-def test_dbt_core_reload_columns_false(
+def test_dbt_core_preserve_metadata(
     mocker: MockerFixture,
     fs: FakeFilesystem,
 ) -> None:
     """
-    Test the ``dbt-core`` command with --preserve-columns flag
+    Test the ``dbt-core`` command with ``--preserve-metadata`` flag.
     """
     root = Path("/path/to/root")
     fs.create_dir(root)
@@ -267,6 +268,76 @@ def test_dbt_core_reload_columns_false(
             str(profiles),
             "--exposures",
             str(exposures),
+            "--preserve-metadata",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0
+    sync_database.assert_called_with(
+        client,
+        profiles,
+        "default",
+        "default",
+        None,
+        False,
+        False,
+        "",
+    )
+
+    sync_datasets.assert_called_with(
+        client,
+        dbt_core_models,
+        dbt_core_metrics,
+        sync_database(),
+        False,
+        "",
+        reload_columns=False,
+        merge_metadata=False,
+    )
+    sync_exposures.assert_called_with(
+        client,
+        exposures,
+        sync_datasets(),
+        {("public", "messages_channels"): "ref('messages_channels')"},
+    )
+
+
+def test_dbt_core_preserve_columns(
+    mocker: MockerFixture,
+    fs: FakeFilesystem,
+) -> None:
+    """
+    Test the ``dbt-core`` command with ``--preserve-columns`` flag.
+    """
+    root = Path("/path/to/root")
+    fs.create_dir(root)
+    manifest = root / "default/target/manifest.json"
+    fs.create_file(manifest, contents=manifest_contents)
+    profiles = root / ".dbt/profiles.yml"
+    fs.create_file(profiles)
+
+    SupersetClient = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.SupersetClient",
+    )
+    client = SupersetClient()
+    mocker.patch("preset_cli.cli.superset.main.UsernamePasswordAuth")
+    sync_database = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.sync_database",
+    )
+    sync_datasets = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.sync_datasets",
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(
+        superset_cli,
+        [
+            "https://superset.example.org/",
+            "sync",
+            "dbt-core",
+            str(manifest),
+            "--profiles",
+            str(profiles),
             "--preserve-columns",
         ],
         catch_exceptions=False,
@@ -291,6 +362,78 @@ def test_dbt_core_reload_columns_false(
         False,
         "",
         reload_columns=False,
+        merge_metadata=False,
+    )
+
+
+def test_dbt_core_merge_metadata(
+    mocker: MockerFixture,
+    fs: FakeFilesystem,
+) -> None:
+    """
+    Test the ``dbt-core`` command with ``--merge-metadata`` flag.
+    """
+    root = Path("/path/to/root")
+    fs.create_dir(root)
+    manifest = root / "default/target/manifest.json"
+    fs.create_file(manifest, contents=manifest_contents)
+    profiles = root / ".dbt/profiles.yml"
+    fs.create_file(profiles)
+    exposures = root / "models/exposures.yml"
+    fs.create_file(exposures)
+
+    SupersetClient = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.SupersetClient",
+    )
+    client = SupersetClient()
+    mocker.patch("preset_cli.cli.superset.main.UsernamePasswordAuth")
+    sync_database = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.sync_database",
+    )
+    sync_datasets = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.sync_datasets",
+    )
+    sync_exposures = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.sync_exposures",
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(
+        superset_cli,
+        [
+            "https://superset.example.org/",
+            "sync",
+            "dbt-core",
+            str(manifest),
+            "--profiles",
+            str(profiles),
+            "--exposures",
+            str(exposures),
+            "--merge-metadata",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0
+    sync_database.assert_called_with(
+        client,
+        profiles,
+        "default",
+        "default",
+        None,
+        False,
+        False,
+        "",
+    )
+
+    sync_datasets.assert_called_with(
+        client,
+        dbt_core_models,
+        dbt_core_metrics,
+        sync_database(),
+        False,
+        "",
+        reload_columns=False,
+        merge_metadata=True,
     )
     sync_exposures.assert_called_with(
         client,
@@ -300,6 +443,45 @@ def test_dbt_core_reload_columns_false(
     )
 
 
+def test_dbt_core_preserve_and_merge(
+    mocker: MockerFixture,
+    fs: FakeFilesystem,
+) -> None:
+    """
+    Test the ``dbt-core`` command with both
+    the ``--preserve-metadata`` and ``--merge-metadata`` flags.
+    """
+    root = Path("/path/to/root")
+    fs.create_dir(root)
+    manifest = root / "default/target/manifest.json"
+    fs.create_file(manifest, contents=manifest_contents)
+    profiles = root / ".dbt/profiles.yml"
+    fs.create_file(profiles)
+    exposures = root / "models/exposures.yml"
+    fs.create_file(exposures)
+
+    mocker.patch("preset_cli.cli.superset.main.UsernamePasswordAuth")
+
+    runner = CliRunner()
+    result = runner.invoke(
+        superset_cli,
+        [
+            "https://superset.example.org/",
+            "sync",
+            "dbt-core",
+            str(manifest),
+            "--profiles",
+            str(profiles),
+            "--exposures",
+            str(exposures),
+            "--preserve-metadata",
+            "--merge-metadata",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 1
+
+
 def test_dbt_core_dbt_project(mocker: MockerFixture, fs: FakeFilesystem) -> None:
     """
     Test the ``dbt-core`` command with a ``dbt_project.yml`` file.
@@ -383,7 +565,10 @@ def test_dbt_core_invalid_argument(mocker: MockerFixture, fs: FakeFilesystem) ->
         catch_exceptions=False,
     )
     assert result.exit_code == 1
-    assert result.output == "FILE should be either manifest.json or dbt_project.yml\n"
+    assert (
+        result.output
+        == "FILE should be either ``manifest.json`` or ``dbt_project.yml``\n"
+    )
 
 
 def test_dbt(mocker: MockerFixture, fs: FakeFilesystem) -> None:
@@ -550,6 +735,7 @@ def test_dbt(mocker: MockerFixture, fs: FakeFilesystem) -> None:
         False,
         "",
         reload_columns=True,
+        merge_metadata=False,
     )
     sync_exposures.assert_called_with(
         client,
@@ -684,7 +870,7 @@ def test_dbt_core_no_database(mocker: MockerFixture, fs: FakeFilesystem) -> None
         catch_exceptions=False,
     )
     assert result.exit_code == 0
-    assert result.output == "No database was found, pass --import-db to create\n"
+    assert result.output == "No database was found, pass ``--import-db`` to create\n"
 
 
 def test_dbt_core_disallow_edits_superset(
@@ -802,12 +988,62 @@ def test_dbt_cloud(mocker: MockerFixture) -> None:
         False,
         "",
         reload_columns=True,
+        merge_metadata=False,
     )
 
 
-def test_dbt_cloud_reload_columns_false(mocker: MockerFixture) -> None:
+def test_dbt_cloud_preserve_metadata(mocker: MockerFixture) -> None:
     """
-    Test the ``dbt-cloud`` command with the --preserve-columns flag.
+    Test the ``dbt-cloud`` command with the ``--preserve-metadata`` flag.
+    """
+    SupersetClient = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.SupersetClient",
+    )
+    superset_client = SupersetClient()
+    mocker.patch("preset_cli.cli.superset.main.UsernamePasswordAuth")
+    DBTClient = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.DBTClient",
+    )
+    dbt_client = DBTClient()
+    sync_datasets = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.sync_datasets",
+    )
+
+    dbt_client.get_models.return_value = dbt_cloud_models
+    dbt_client.get_metrics.return_value = dbt_cloud_metrics
+    database = mocker.MagicMock()
+    superset_client.get_databases.return_value = [database]
+    superset_client.get_database.return_value = database
+
+    runner = CliRunner()
+    result = runner.invoke(
+        superset_cli,
+        [
+            "https://superset.example.org/",
+            "sync",
+            "dbt-cloud",
+            "XXX",
+            "123",
+            "--preserve-metadata",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0
+    sync_datasets.assert_called_with(
+        superset_client,
+        dbt_cloud_models,
+        dbt_cloud_metrics,
+        database,
+        False,
+        "",
+        reload_columns=False,
+        merge_metadata=False,
+    )
+
+
+def test_dbt_cloud_preserve_columns(mocker: MockerFixture) -> None:
+    """
+    Test the ``dbt-cloud`` command with the ``--preserve-columns`` flag.
     """
     SupersetClient = mocker.patch(
         "preset_cli.cli.superset.sync.dbt.command.SupersetClient",
@@ -850,7 +1086,81 @@ def test_dbt_cloud_reload_columns_false(mocker: MockerFixture) -> None:
         False,
         "",
         reload_columns=False,
+        merge_metadata=False,
+    )
+
+
+def test_dbt_cloud_merge_metadata(mocker: MockerFixture) -> None:
+    """
+    Test the ``dbt-cloud`` command with the ``--merge-metadata`` flag.
+    """
+    SupersetClient = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.SupersetClient",
     )
+    superset_client = SupersetClient()
+    mocker.patch("preset_cli.cli.superset.main.UsernamePasswordAuth")
+    DBTClient = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.DBTClient",
+    )
+    dbt_client = DBTClient()
+    sync_datasets = mocker.patch(
+        "preset_cli.cli.superset.sync.dbt.command.sync_datasets",
+    )
+
+    dbt_client.get_models.return_value = dbt_cloud_models
+    dbt_client.get_metrics.return_value = dbt_cloud_metrics
+    database = mocker.MagicMock()
+    superset_client.get_databases.return_value = [database]
+    superset_client.get_database.return_value = database
+
+    runner = CliRunner()
+    result = runner.invoke(
+        superset_cli,
+        [
+            "https://superset.example.org/",
+            "sync",
+            "dbt-cloud",
+            "XXX",
+            "123",
+            "--merge-metadata",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0
+    sync_datasets.assert_called_with(
+        superset_client,
+        dbt_cloud_models,
+        dbt_cloud_metrics,
+        database,
+        False,
+        "",
+        reload_columns=False,
+        merge_metadata=True,
+    )
+
+
+def test_dbt_cloud_preserve_and_merge(mocker: MockerFixture) -> None:
+    """
+    Test the ``dbt-cloud`` command with both
+    the ``--preserve-metadata`` and ``--merge-metadata`` flags.
+    """
+    mocker.patch("preset_cli.cli.superset.main.UsernamePasswordAuth")
+
+    runner = CliRunner()
+    result = runner.invoke(
+        superset_cli,
+        [
+            "https://superset.example.org/",
+            "sync",
+            "dbt-cloud",
+            "XXX",
+            "123",
+            "--preserve-metadata",
+            "--merge-metadata",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 1
 
 
 def test_dbt_cloud_no_job_id(mocker: MockerFixture) -> None:
@@ -902,6 +1212,7 @@ def test_dbt_cloud_no_job_id(mocker: MockerFixture) -> None:
         False,
         "",
         reload_columns=True,
+        merge_metadata=False,
     )
 
 
diff --git a/tests/cli/superset/sync/dbt/databases_test.py b/tests/cli/superset/sync/dbt/databases_test.py
index 0ac22875..d7bf4180 100644
--- a/tests/cli/superset/sync/dbt/databases_test.py
+++ b/tests/cli/superset/sync/dbt/databases_test.py
@@ -397,7 +397,7 @@ def test_sync_database_reuse_connection(
     fs: FakeFilesystem,
 ) -> None:
     """
-    Test ``sync_database`` when the connection already exists and --import-db wasn't passed.
+    Test ``sync_database`` when the connection already exists and ``--import-db`` wasn't passed.
     """
     fs.create_file(
         "/path/to/.dbt/profiles.yml",
diff --git a/tests/cli/superset/sync/dbt/datasets_test.py b/tests/cli/superset/sync/dbt/datasets_test.py
index ede20d5a..b88fc18b 100644
--- a/tests/cli/superset/sync/dbt/datasets_test.py
+++ b/tests/cli/superset/sync/dbt/datasets_test.py
@@ -3,6 +3,7 @@
 """
 # pylint: disable=invalid-name, too-many-lines
 
+import copy
 import json
 from typing import List, cast
 from unittest import mock
@@ -515,11 +516,28 @@ def test_sync_datasets_external_url_disallow_edits(mocker: MockerFixture) -> Non
     )
 
 
-def test_sync_datasets_preserve_columns(mocker: MockerFixture) -> None:
+def test_sync_datasets_preserve_metadata(mocker: MockerFixture) -> None:
     """
-    Test ``sync_datasets`` when setting ovverride_columns to false.
+    Test ``sync_datasets`` with preserve_metadata set to True.
+    Metrics should be merged (Preset as the source of truth).
     """
     client = mocker.MagicMock()
+    metrics_ = copy.deepcopy(metrics)
+    metrics_.append(
+        metric_schema.load(
+            {
+                "depends_on": ["model.superset_examples.messages_channels"],
+                "description": "",
+                "filters": [],
+                "meta": {},
+                "name": "max_id",
+                "label": "",
+                "sql": "id",
+                "type": "max",
+                "unique_id": "metric.superset_examples.cnt",
+            },
+        ),
+    )
     client.get_datasets.side_effect = [[{"id": 1}], [{"id": 2}], [{"id": 3}]]
     client.get_dataset.return_value = {
         "columns": [
@@ -530,12 +548,29 @@ def test_sync_datasets_preserve_columns(mocker: MockerFixture) -> None:
                 "groupby": False,
             },
         ],
+        "metrics": [
+            {
+                "changed_on": "2023-08-28T18:01:24.190211",
+                "created_on": "2023-08-28T18:01:24.190208",
+                "expression": "count(*)/1",
+                "id": 190,
+                "metric_name": "cnt",
+            },
+            {
+                "changed_on": "2023-08-27T18:01:24.190211",
+                "created_on": "2023-08-27T18:01:24.190208",
+                "expression": "COUNT (DISTINCT id)",
+                "id": 191,
+                "metric_name": "unique_ids",
+            },
+        ],
+        "id": 1,
     }
 
     sync_datasets(
         client=client,
         models=models,
-        metrics=metrics,
+        metrics=metrics_,
         database={"id": 1},
         disallow_edits=False,
         external_url_prefix="https://dbt.example.org/",
@@ -556,7 +591,26 @@ def test_sync_datasets_preserve_columns(mocker: MockerFixture) -> None:
                     },
                 ),
                 is_managed_externally=False,
-                metrics=[],
+                metrics=[
+                    {
+                        "expression": "count(*)/1",
+                        "id": 190,
+                        "metric_name": "cnt",
+                    },
+                    {
+                        "expression": "COUNT (DISTINCT id)",
+                        "id": 191,
+                        "metric_name": "unique_ids",
+                    },
+                    {
+                        "expression": "MAX(id)",
+                        "metric_name": "max_id",
+                        "metric_type": "max",
+                        "verbose_name": "",
+                        "description": "",
+                        "extra": "{}",
+                    },
+                ],
                 external_url=(
                     "https://dbt.example.org/"
                     "#!/model/model.superset_examples.messages_channels"
@@ -565,7 +619,103 @@ def test_sync_datasets_preserve_columns(mocker: MockerFixture) -> None:
             mock.call(
                 1,
                 override_columns=False,
+                columns=[
+                    {
+                        "column_name": "id",
+                        "description": "Primary key",
+                        "filterable": False,
+                        "groupby": False,
+                        "is_dttm": False,
+                        "verbose_name": "id",
+                    },
+                ],
+            ),
+        ],
+    )
+
+
+def test_sync_datasets_merge_metadata(mocker: MockerFixture) -> None:
+    """
+    Test ``sync_datasets`` with merge_metadata set to True.
+    Metrics should be merged (dbt as the source of truth).
+    """
+    client = mocker.MagicMock()
+    metrics_ = copy.deepcopy(metrics)
+    metrics_.append(
+        metric_schema.load(
+            {
+                "depends_on": ["model.superset_examples.messages_channels"],
+                "description": "",
+                "filters": [],
+                "meta": {},
+                "name": "max_id",
+                "label": "",
+                "sql": "id",
+                "type": "max",
+                "unique_id": "metric.superset_examples.cnt",
+            },
+        ),
+    )
+    client.get_datasets.side_effect = [[{"id": 1}], [{"id": 2}], [{"id": 3}]]
+    client.get_dataset.return_value = {
+        "columns": [
+            {
+                "column_name": "id",
+                "is_dttm": False,
+                "filterable": False,
+                "groupby": False,
+            },
+        ],
+        "metrics": [
+            {
+                "changed_on": "2023-08-28T18:01:24.190211",
+                "created_on": "2023-08-28T18:01:24.190208",
+                "expression": "count(*)/1",
+                "id": 190,
+                "metric_name": "cnt",
+            },
+            {
+                "changed_on": "2023-08-27T18:01:24.190211",
+                "created_on": "2023-08-27T18:01:24.190208",
+                "expression": "COUNT (DISTINCT id)",
+                "id": 191,
+                "metric_name": "unique_ids",
+            },
+        ],
+        "id": 1,
+    }
+
+    sync_datasets(
+        client=client,
+        models=models,
+        metrics=metrics_,
+        database={"id": 1},
+        disallow_edits=False,
+        external_url_prefix="https://dbt.example.org/",
+        reload_columns=False,
+        merge_metadata=True,
+    )
+    client.create_dataset.assert_not_called()
+    client.update_dataset.assert_has_calls(
+        [
+            mock.call(
+                1,
+                override_columns=False,
+                description="",
+                extra=json.dumps(
+                    {
+                        "unique_id": "model.superset_examples.messages_channels",
+                        "depends_on": "ref('messages_channels')",
+                        "certification": {"details": "This table is produced by dbt"},
+                    },
+                ),
+                is_managed_externally=False,
                 metrics=[
+                    {
+                        "expression": "COUNT (DISTINCT id)",
+                        "id": 191,
+                        "metric_name": "unique_ids",
+                    },
                     {
                         "expression": "COUNT(*)",
                         "metric_name": "cnt",
@@ -573,8 +723,21 @@ def test_sync_datasets_preserve_columns(mocker: MockerFixture) -> None:
                         "verbose_name": "",
                         "description": "",
                         "extra": "{}",
+                        "id": 190,
+                    },
+                    {
+                        "expression": "MAX(id)",
+                        "metric_name": "max_id",
+                        "metric_type": "max",
+                        "verbose_name": "",
+                        "description": "",
+                        "extra": "{}",
                     },
                 ],
+                external_url=(
+                    "https://dbt.example.org/"
+                    "#!/model/model.superset_examples.messages_channels"
+                ),
             ),
             mock.call(
                 1,
@@ -583,9 +746,9 @@ def test_sync_datasets_preserve_columns(mocker: MockerFixture) -> None:
                     {
                         "column_name": "id",
                         "description": "Primary key",
-                        "is_dttm": False,
                         "filterable": False,
                         "groupby": False,
+                        "is_dttm": False,
                         "verbose_name": "id",
                     },
                 ],
diff --git a/tests/cli/superset/sync/native/command_test.py b/tests/cli/superset/sync/native/command_test.py
index c5bb44ba..18baf306 100644
--- a/tests/cli/superset/sync/native/command_test.py
+++ b/tests/cli/superset/sync/native/command_test.py
@@ -108,7 +108,7 @@ def test_import_resources_overwrite_needed(mocker: MockerFixture) -> None:
     import_resources(contents=contents, client=client, overwrite=False)
 
     assert click.style.called_with(
-        "The following file(s) already exist. Pass --overwrite to replace them.\n"
+        "The following file(s) already exist. Pass ``--overwrite`` to replace them.\n"
         "databases/gsheets.yaml",
         fg="bright_red",
     )
@@ -814,7 +814,7 @@ def test_sync_native_jinja_templating_disabled(
     fs: FakeFilesystem,
 ) -> None:
     """
-    Test ``native`` command with --disable-jinja-templating.
+    Test ``native`` command with ``--disable-jinja-templating``.
     """
     root = Path("/path/to/root")
     fs.create_dir(root)