@@ -110,7 +110,7 @@ doris_ready() {
     if [ "$HOSTNAME" = "localhost" ]; then
         HOSTNAME="127.0.0.1"
     fi
-    INIT_SQL="ADMIN SET FRONTEND CONFIG ('default_replication_num' = '1'); ADMIN SET FRONTEND CONFIG ('force_olap_table_replication_num' = '1'); ADMIN SET FRONTEND CONFIG ('enable_unicode_name_support' = 'true');"
+    INIT_SQL="ADMIN SET FRONTEND CONFIG ('force_olap_table_replication_num' = '1');"
     if docker ps --format '{{.Names}}' 2>/dev/null | grep -qx 'doris-fe-01'; then
         docker exec -i doris-fe-01 mysql -h127.0.0.1 -P9030 -uroot -e "SHOW BACKENDS\G" 2>/dev/null | grep -q "Alive: true" && \
         docker exec -i doris-fe-01 mysql -h127.0.0.1 -P9030 -uroot -e "$INIT_SQL" 2>/dev/null
@@ -128,9 +128,7 @@ with conn.cursor() as cur:
     alive_idx = cols.index('Alive')
     if not rows or not any(str(r[alive_idx]).lower() == 'true' for r in rows):
         sys.exit(1)
-    for cfg in ('default_replication_num', 'force_olap_table_replication_num'):
-        cur.execute(f\"ADMIN SET FRONTEND CONFIG ('{cfg}' = '1')\")
-    cur.execute(\"ADMIN SET FRONTEND CONFIG ('enable_unicode_name_support' = 'true')\")
+    cur.execute(\"ADMIN SET FRONTEND CONFIG ('force_olap_table_replication_num' = '1')\")
 " 2>/dev/null
     fi
 }
 
@@ -2851,6 +2851,9 @@ def init(cursor: t.Any) -> None:
 
         return init
 
+    def get_catalog(self) -> t.Optional[str]:
+        return self.catalog
+
 
 CONNECTION_CONFIG_TO_TYPE = {
     # Map all subclasses of ConnectionConfig to the value of their `type_` field.
 
@@ -891,10 +891,19 @@ def _extract_subquery_from_where(
         else:
             key_exprs = [lhs.copy()]
 
-        if not key_exprs or not all(isinstance(k, exp.Column) for k in key_exprs):
+        normalized_keys: t.List[exp.Expr] = []
+        for k in key_exprs:
+            if isinstance(k, exp.Column):
+                normalized_keys.append(k)
+            elif isinstance(k, exp.Identifier):
+                normalized_keys.append(exp.Column(this=k.copy()))
+            else:
+                return None
+
+        if not normalized_keys:
             return None
 
-        return key_exprs, subquery, extra_conditions
+        return normalized_keys, subquery, extra_conditions
 
     @staticmethod
     def _unwrap_concat_ws(expr: exp.Expr) -> t.Optional[exp.Expr]:
 
@@ -440,6 +440,12 @@ def test_materialized_view(ctx_query_and_df: TestContext):
             "which this generic test does not provide; StarRocks MVs are covered by "
             "test_integration_starrocks.py"
         )
+    if ctx.engine_adapter.dialect == "doris":
+        pytest.skip(
+            "Doris materialized views require engine-specific properties (SYNC source_table or MTMV DISTRIBUTED BY), "
+            "which this generic test does not provide; Doris MVs are covered by "
+            "test_integration_doris.py"
+        )
     input_data = pd.DataFrame(
         [
             {"id": 1, "ds": "2022-01-01"},
@@ -450,22 +456,8 @@ def test_materialized_view(ctx_query_and_df: TestContext):
     source_table = ctx.table("source_table")
     ctx.engine_adapter.ctas(source_table, ctx.input_data(input_data), ctx.columns_to_types)
     view = ctx.table("test_view")
-    # For Doris synchronized materialized views, we need to use only the table name without schema and add the source table name to the table properties
-    if ctx.engine_adapter.dialect == "doris":
-        view = exp.to_table(view.name)
-        # For Doris, select only the first column name from columns_to_types.
-        # Otherwise, Doris will raise an error if the materialized view has the same schema as the base table.
-        first_column = next(iter(ctx.columns_to_types))
-        view_query = exp.select(first_column).from_(source_table)
-        ctx.engine_adapter.create_view(
-            view,
-            view_query,
-            materialized=True,
-            view_properties={"materialized_type": "SYNC", "source_table": source_table},
-        )
-    else:
-        view_query = exp.select(*ctx.columns_to_types).from_(source_table)
-        ctx.engine_adapter.create_view(view, view_query, materialized=True)
+    view_query = exp.select(*ctx.columns_to_types).from_(source_table)
+    ctx.engine_adapter.create_view(view, view_query, materialized=True)
     results = ctx.get_metadata_results()
     # Redshift considers the underlying dataset supporting materialized views as a table therefore we get 2
     # tables in the result
@@ -474,31 +466,13 @@ def test_materialized_view(ctx_query_and_df: TestContext):
     else:
         assert len(results.tables) == 1
     assert len(results.views) == 0
-    # Doris information_schema.tables does not show materialized views and cannot query the synchronized materialized view
-    if ctx.engine_adapter.dialect != "doris":
-        assert len(results.materialized_views) == 1
-        assert results.materialized_views[0] == view.name
-        ctx.compare_with_current(view, input_data)
+    assert len(results.materialized_views) == 1
+    assert results.materialized_views[0] == view.name
+    ctx.compare_with_current(view, input_data)
     # Make sure that dropping a materialized view also works
-    if ctx.engine_adapter.dialect == "doris":
-        # Wait for the materialized view to be created by retrying drop until it succeeds
-        def drop_view_success():
-            try:
-                ctx.engine_adapter.drop_view(
-                    view,
-                    ignore_if_not_exists=False,
-                    materialized=True,
-                    view_properties={"materialized_type": "SYNC", "source_table": source_table},
-                )
-                return True
-            except Exception:
-                return False
-
-        wait_until(drop_view_success, attempts=5, wait=1)
-    else:
-        ctx.engine_adapter.drop_view(view, materialized=True)
-        results = ctx.get_metadata_results()
-        assert len(results.materialized_views) == 0
+    ctx.engine_adapter.drop_view(view, materialized=True)
+    results = ctx.get_metadata_results()
+    assert len(results.materialized_views) == 0
 
 
 def test_drop_schema(ctx: TestContext):
@@ -1031,12 +1005,8 @@ def test_merge(ctx_query_and_df: TestContext):
     # Athena only supports MERGE on Iceberg tables
     # And it cant fall back to a logical merge on Hive tables because it cant delete records
     table_format = "iceberg" if ctx.dialect == "athena" else None
-    # Doris needs to have a UNIQUE KEY to use MERGE
-    table_properties = {"unique_key": "id"} if ctx.dialect == "doris" else None
 
-    ctx.engine_adapter.create_table(
-        table, ctx.columns_to_types, table_format=table_format, table_properties=table_properties
-    )
+    ctx.engine_adapter.create_table(table, ctx.columns_to_types, table_format=table_format)
     input_data = pd.DataFrame(
         [
             {"id": 1, "ds": "2022-01-01"},
@@ -1101,15 +1071,11 @@ def test_merge_source_columns(ctx_query_and_df: TestContext):
     # Athena only supports MERGE on Iceberg tables
     # And it cant fall back to a logical merge on Hive tables because it cant delete records
     table_format = "iceberg" if ctx.dialect == "athena" else None
-    # Doris needs to have a UNIQUE KEY to use MERGE
-    table_properties = {"unique_key": "id"} if ctx.dialect == "doris" else None
 
     columns_to_types = ctx.columns_to_types.copy()
     columns_to_types["ignored_column"] = exp.DataType.build("int")
 
-    ctx.engine_adapter.create_table(
-        table, columns_to_types, table_format=table_format, table_properties=table_properties
-    )
+    ctx.engine_adapter.create_table(table, columns_to_types, table_format=table_format)
     input_data = pd.DataFrame(
         [
             {"id": 1, "ds": "2022-01-01", "ignored_source": "ignored_value"},
@@ -4058,7 +4024,7 @@ def test_unicode_characters(ctx: TestContext, tmp_path: Path):
     # I also think Spark may not support unicode in general but that would need to be verified.
     if not ctx.engine_adapter.QUOTE_IDENTIFIERS_IN_VIEWS:
         pytest.skip("Skipping as these engines have issues with unicode characters in model names")
-    # Doris default setting `enable_unicode_name_support=false` so it is incompatible with unicode characters in model names
+    # Doris static fe.conf setting `enable_unicode_name_support` defaults to false (and is not runtime-mutable)
     if ctx.dialect == "doris":
         pytest.skip(
             "Skipping as Doris default setting has issues with unicode characters in model names"
 
@@ -98,3 +98,125 @@ def test_doris_iceberg_local_catalog(ctx: TestContext):
         except Exception:
             pass
         adapter.set_current_catalog("internal")
+
+
+def test_materialized_view_sync(ctx: TestContext):
+    import pandas as pd  # noqa: TID253
+    from tests.core.engine_adapter.integration import wait_until
+
+    input_data = pd.DataFrame(
+        [
+            {"id": 1, "ds": "2022-01-01"},
+            {"id": 2, "ds": "2022-01-02"},
+            {"id": 3, "ds": "2022-01-03"},
+        ]
+    )
+    source_table = ctx.table("source_table")
+    ctx.engine_adapter.ctas(source_table, ctx.input_data(input_data), ctx.columns_to_types)
+    view = exp.to_table("test_sync_mv")
+    first_column = next(iter(ctx.columns_to_types))
+    view_query = exp.select(first_column).from_(source_table)
+    ctx.engine_adapter.create_view(
+        view,
+        view_query,
+        materialized=True,
+        view_properties={
+            "materialized_type": exp.Literal.string("SYNC"),
+            "source_table": source_table,
+        },
+    )
+
+    def drop_view_success() -> bool:
+        try:
+            ctx.engine_adapter.drop_view(
+                view,
+                ignore_if_not_exists=False,
+                materialized=True,
+                view_properties={
+                    "materialized_type": exp.Literal.string("SYNC"),
+                    "source_table": source_table,
+                },
+            )
+            return True
+        except Exception:
+            return False
+
+    wait_until(drop_view_success, attempts=5, wait=1)
+
+
+def test_materialized_view_async_mtmv(ctx: TestContext):
+    import pandas as pd  # noqa: TID253
+    from tests.core.engine_adapter.integration import wait_until
+
+    input_data = pd.DataFrame(
+        [
+            {"id": 1, "ds": "2022-01-01"},
+            {"id": 2, "ds": "2022-01-02"},
+            {"id": 3, "ds": "2022-01-03"},
+        ]
+    )
+    source_table = ctx.table("mtmv_source")
+    ctx.engine_adapter.ctas(source_table, ctx.input_data(input_data), ctx.columns_to_types)
+    view = ctx.table("test_async_mtmv")
+    view_query = exp.select(*ctx.columns_to_types).from_(source_table)
+    ctx.engine_adapter.create_view(
+        view,
+        view_query,
+        materialized=True,
+        view_properties={
+            "build": exp.Literal.string("IMMEDIATE"),
+            "refresh": exp.Literal.string("COMPLETE"),
+            "refresh_trigger": exp.Literal.string("ON MANUAL"),
+            "distributed_by": exp.Literal.string("RANDOM BUCKETS AUTO"),
+        },
+    )
+    wait_until(lambda: len(ctx.get_current_data(view)) == len(input_data), attempts=10, wait=1)
+    ctx.compare_with_current(view, input_data)
+    ctx.engine_adapter.drop_view(view, materialized=True)
+
+
+def test_merge_unique_key(ctx: TestContext):
+    import pandas as pd  # noqa: TID253
+
+    table = ctx.table("test_merge_uk")
+    ctx.engine_adapter.create_table(
+        table,
+        ctx.columns_to_types,
+        table_properties={"unique_key": exp.Literal.string("id")},
+    )
+    input_data = pd.DataFrame(
+        [
+            {"id": 1, "ds": "2022-01-01"},
+            {"id": 2, "ds": "2022-01-02"},
+            {"id": 3, "ds": "2022-01-03"},
+        ]
+    )
+    ctx.engine_adapter.merge(
+        table,
+        ctx.input_data(input_data),
+        target_columns_to_types=None,
+        unique_key=[exp.to_identifier("id")],
+    )
+    merge_data = pd.DataFrame(
+        [
+            {"id": 2, "ds": "2022-01-10"},
+            {"id": 4, "ds": "2022-01-04"},
+        ]
+    )
+    ctx.engine_adapter.merge(
+        table,
+        ctx.input_data(merge_data),
+        target_columns_to_types=None,
+        unique_key=[exp.to_identifier("id")],
+    )
+    ctx.compare_with_current(
+        table,
+        pd.DataFrame(
+            [
+                {"id": 1, "ds": "2022-01-01"},
+                {"id": 2, "ds": "2022-01-10"},
+                {"id": 3, "ds": "2022-01-03"},
+                {"id": 4, "ds": "2022-01-04"},
+            ]
+        ),
+    )
@@ -1277,3 +1277,27 @@ def test_adjust_physical_properties_for_incremental(
     assert to_sql_calls(adapter) == [
         "CREATE TABLE IF NOT EXISTS `db`.`incremental_model` (`id` INT, `name` VARCHAR(50)) UNIQUE KEY (`id`) DISTRIBUTED BY HASH (`id`) BUCKETS 10"
     ]
+
+
+def test_merge_with_identifier_unique_key(
+    make_mocked_engine_adapter: t.Callable[..., DorisEngineAdapter],
+    mocker: MockerFixture,
+):
+    adapter = make_mocked_engine_adapter(DorisEngineAdapter)
+    mocker.patch("sqlmesh.core.engine_adapter.doris.random_id", return_value="abcd1234")
+    adapter.merge(
+        target_table="db.target",
+        source_table=parse_one("SELECT id, ds FROM db.src"),
+        target_columns_to_types={
+            "id": exp.DataType.build("INT", dialect="doris"),
+            "ds": exp.DataType.build("VARCHAR(32)", dialect="doris"),
+        },
+        unique_key=[exp.to_identifier("id")],
+    )
+    sql_calls = to_sql_calls(adapter)
+    assert any(
+        call.startswith(
+            "DELETE FROM `db`.`target` AS `_t1` USING (SELECT `id` FROM `db`.`__temp_target_abcd1234`) AS `_t2` WHERE `_t1`.`id` = `_t2`.`id`"
+        )
+        for call in sql_calls
+    )