fix(doris): preserve table alias on self-referencing replace_query, wait on index SCHEMA_CHANGE, and refine Lance TVF docs

Signed-off-by: Bruce Arctor <5032356+brucearctor@users.noreply.github.com>

Author: brucearctor

docs/integrations/engines/doris.md

@@ -467,7 +467,7 @@ FROM hive_catalog.raw.orders;
 
 ### Pattern 5: Reading External Lance, Hudi & Delta Lake Tables (Doris 4.1+)
 
-Apache Doris 4.1+ supports reading multimodal and vector datasets stored in **Lance** format (`'type' = 'lance'` catalog or `s3(...)` / `gcs(...)` / `hdfs(...)` / `local(...)` table-valued functions with `'format' = 'lance'`), as well as external **Apache Hudi** (`'type' = 'hudi'`) and **Delta Lake** (`'type' = 'deltalake'`) catalogs.
+Apache Doris 4.1.4+ supports reading multimodal and vector datasets stored in **Lance** format (`'type' = 'lance'` catalog or `s3(...)` / `local(...)` table-valued functions with `'format' = 'lance'`), as well as external **Apache Hudi** (`'type' = 'hudi'`) and **Delta Lake** (`'type' = 'deltalake'`) catalogs.
 
 ```yaml
 # config.yaml
@@ -773,7 +773,7 @@ MODEL (
 
 | SQLMesh Model Kind | Doris Internal OLAP Catalog (`internal` on Local Disk or S3/GCS Storage Vault) | External Iceberg & Paimon Catalogs (`table_format iceberg \| paimon` on S3/GCS/HDFS) | External Hive Catalog (`table_format hive` on S3/GCS/HDFS) | External Read-Only Lakehouse Catalogs (`lance`, `hudi`, `deltalake`, `lakesoul`) |
 | --- | :---: | :---: | :---: | :---: |
-| Source / Upstream `SELECT` | Supported | Supported (incl. `FOR TIME/VERSION AS OF`, `@branch`, `@tag`) | Supported | Supported (Read-Only via Catalog or S3/GCS TVF) |
+| Source / Upstream `SELECT` | Supported | Supported (incl. `FOR TIME/VERSION AS OF`, `@branch`, `@tag`) | Supported | Supported (Read-Only via corresponding External Catalog; Lance also supports `s3()` / `local()` TVFs) |
 | `FULL` | Supported (`CREATE TABLE ... AS SELECT` / `DELETE` + `INSERT`) | Supported (`CREATE TABLE ... ENGINE=iceberg\|paimon` / `INSERT OVERWRITE TABLE`) | Supported (`CREATE TABLE ... ENGINE=hive` / `INSERT OVERWRITE TABLE`) | Read-Only in Doris (Write Guarded) |
 | `INCREMENTAL_BY_TIME_RANGE` | Supported (`DELETE WHERE time_col >= low AND time_col <= high` + `INSERT`) | Supported (`DELETE` + `INSERT`; requires Doris 4.1.0+ & Iceberg v2 / Paimon PK) | Not supported by Hive in Doris (No row-level `DELETE`) | Read-Only in Doris (Write Guarded) |
 | `INCREMENTAL_BY_UNIQUE_KEY` | Supported (Auto-promotes `unique_key` to `UNIQUE KEY`, `DELETE ... USING` + `INSERT` or native `MERGE INTO` when `when_matched` / `merge_filter` is specified) | Supported (Batched key `DELETE` + `INSERT` or native `MERGE INTO`; requires Doris 4.1.0+ & Iceberg v2 / Paimon PK) | Not supported by Hive in Doris (No `DELETE` / `MERGE INTO`) | Read-Only in Doris (Write Guarded) |

sqlmesh/core/engine_adapter/base.py

@@ -531,6 +531,11 @@ def replace_query(
         default_catalog_loaded = False
         qualified_target: t.Optional[exp.Table] = None
 
+        def _table_parts(
+            table: exp.Table,
+        ) -> t.Tuple[t.Optional[exp.Expr], t.Optional[exp.Expr], t.Optional[exp.Expr]]:
+            return (table.args.get("catalog"), table.args.get("db"), table.this)
+
         def _qualified(table: exp.Table, catalog: str) -> exp.Table:
             table = table.copy()
             if not table.catalog and table.db:
@@ -542,7 +547,7 @@ def _is_target_table(node: exp.Expr) -> bool:
             if not isinstance(node, exp.Table):
                 return False
             quoted_node = quote_identifiers(node.copy())
-            if quoted_node == quoted_target:
+            if _table_parts(quoted_node) == _table_parts(quoted_target):
                 return True
             if quoted_node.this != quoted_target.this or quoted_node.args.get(
                 "db"
@@ -555,9 +560,19 @@ def _is_target_table(node: exp.Expr) -> bool:
                     qualified_target = _qualified(target_table, default_catalog)
             return (
                 default_catalog is not None
-                and _qualified(node, default_catalog) == qualified_target
+                and qualified_target is not None
+                and _table_parts(_qualified(node, default_catalog))
+                == _table_parts(qualified_target)
             )
 
+        def _replace_target_table(node: exp.Expr, replacement_table: exp.Table) -> exp.Expr:
+            if not _is_target_table(node):
+                return node
+            replacement = replacement_table.copy()
+            if isinstance(node, exp.Table) and node.args.get("alias") is not None:
+                replacement.set("alias", node.args["alias"].copy())
+            return replacement
+
         self_referencing = any(_is_target_table(table) for table in query.find_all(exp.Table))
         # If a query references itself then it must have a table created regardless of approach used.
         if self_referencing:
@@ -601,9 +616,7 @@ def _is_target_table(node: exp.Expr) -> bool:
             ) as temp_table:
                 for source_query in source_queries:
                     source_query.add_transform(
-                        lambda node: (  # type: ignore
-                            temp_table if _is_target_table(node) else node  # type: ignore
-                        )
+                        lambda node: _replace_target_table(node, temp_table)  # type: ignore
                     )
                 return self._insert_overwrite_by_condition(
                     target_table,

sqlmesh/core/engine_adapter/doris.py

@@ -1644,9 +1644,11 @@ def create_index(
                     exists=exists,
                 )
             )
+            self._wait_for_schema_change(table_name)
             return
         col_names = tuple(c.name if isinstance(c, exp.Expr) else str(c) for c in columns)
         super().create_index(table_name, index_name, col_names, exists=exists)
+        self._wait_for_schema_change(table_name)
 
     def _create_table_like(
         self,
@@ -4098,17 +4100,32 @@ def _create_table_from_source_queries(
         first_query = source_queries[0].query_factory() if source_queries else None
 
         def _matches_target_table(t_node: exp.Table) -> bool:
-            if quote_identifiers(t_node.copy()) == quote_identifiers(table.copy()):
+            q_node = quote_identifiers(t_node.copy())
+            q_target = quote_identifiers(table.copy())
+            if (q_node.args.get("catalog"), q_node.args.get("db"), q_node.this) == (
+                q_target.args.get("catalog"),
+                q_target.args.get("db"),
+                q_target.this,
+            ):
                 return True
-            if self.default_catalog:
-                norm_node = t_node.copy()
-                if not norm_node.catalog and norm_node.db:
-                    norm_node.set("catalog", exp.parse_identifier(self.default_catalog))
-                norm_target = table.copy()
-                if not norm_target.catalog and norm_target.db:
-                    norm_target.set("catalog", exp.parse_identifier(self.default_catalog))
-                return quote_identifiers(norm_node) == quote_identifiers(norm_target)
-            return False
+            default_cat = self.default_catalog or "internal"
+            norm_node = t_node.copy()
+            if not norm_node.catalog and norm_node.db:
+                norm_node.set("catalog", exp.parse_identifier(default_cat))
+            norm_target = table.copy()
+            if not norm_target.catalog and norm_target.db:
+                norm_target.set("catalog", exp.parse_identifier(default_cat))
+            q_norm_node = quote_identifiers(norm_node)
+            q_norm_target = quote_identifiers(norm_target)
+            return (
+                q_norm_node.args.get("catalog"),
+                q_norm_node.args.get("db"),
+                q_norm_node.this,
+            ) == (
+                q_norm_target.args.get("catalog"),
+                q_norm_target.args.get("db"),
+                q_norm_target.this,
+            )
 
         self_referencing = bool(
             first_query

tests/core/engine_adapter/integration/test_integration_doris.py

@@ -872,3 +872,63 @@ def test_doris_4_1_native_merge_segment_v3_and_ann_index(ctx: TestContext):
     )
     seq_rows = doris_adapter.fetchall(exp.select("id", "val", "ver").from_(seq_tbl))
     assert list(seq_rows) == [(1, "v10", 10)]
+
+    # 7. Self-referencing replace_query with aliased, catalog-qualified `internal.<db>.<tbl> AS t`
+    self_ref_tbl = ctx.table("test_doris_self_ref")
+    self_ref_cols = {
+        "id": exp.DataType.build("INT", dialect="doris"),
+        "val": exp.DataType.build("INT", dialect="doris"),
+    }
+    doris_adapter.create_table(self_ref_tbl, self_ref_cols)
+    doris_adapter.execute(
+        f"INSERT INTO {doris_adapter._to_sql(self_ref_tbl)} (`id`, `val`) VALUES (1, 10), (2, 20)"
+    )
+    prev_default_cat = doris_adapter._default_catalog
+    try:
+        doris_adapter._default_catalog = "internal"
+        doris_adapter.replace_query(
+            self_ref_tbl,
+            parse_one(
+                f"SELECT t.id, t.val + 5 AS val FROM `internal`.`{self_ref_tbl.db}`.`{self_ref_tbl.name}` AS t",
+                read="doris",
+            ),
+            target_columns_to_types=self_ref_cols,
+        )
+    finally:
+        doris_adapter._default_catalog = prev_default_cat
+    self_ref_rows = doris_adapter.fetchall(
+        exp.select("id", "val").from_(self_ref_tbl).order_by("id")
+    )
+    assert list(self_ref_rows) == [(1, 15), (2, 25)]
+
+    # 8. NGRAM_BF index creation + BUILD INDEX on INVERTED index + TIMESTAMPTZ cross-timezone normalize_value equivalence
+    doris_adapter.create_index(
+        fts_tbl,
+        "idx_msg_ngram",
+        [exp.to_column("msg")],
+        index_type="NGRAM_BF",
+        index_properties={"gram_size": "3", "bf_size": "256"},
+    )
+    doris_adapter.execute(
+        parse_one(
+            f"BUILD INDEX idx_msg ON {doris_adapter._to_sql(fts_tbl)}",
+            read="doris",
+        )
+    )
+    norm_tz = doris_adapter.normalize_value(
+        exp.to_column("ts"),
+        exp.DataType.build("TIMESTAMPTZ(6)", dialect="doris"),
+        timestamp_precision=6,
+    )
+    tz_check_rows = doris_adapter.fetchall(
+        exp.select(norm_tz.as_("utc_str")).from_(
+            parse_one(
+                "(SELECT CAST('2024-06-01 12:00:00.123456+08:00' AS TIMESTAMPTZ(6)) AS ts "
+                "UNION ALL "
+                "SELECT CAST('2024-06-01 04:00:00.123456+00:00' AS TIMESTAMPTZ(6)) AS ts) t",
+                read="doris",
+            )
+        )
+    )
+    assert len(tz_check_rows) == 2
+    assert tz_check_rows[0][0] == tz_check_rows[1][0] == "2024-06-01 04:00:00.123456"

tests/core/engine_adapter/test_doris.py

@@ -4865,11 +4865,12 @@ def _flaky_execute(sql: str, **kwargs: t.Any) -> None:
     adapter.cursor.execute.side_effect = _flaky_execute
     adapter.alter_table(
         [
-            parse_one(
-                "ALTER TABLE `db`.`t` MODIFY COLUMN `a` BIGINT", into=exp.Alter, read="doris"
+            t.cast(
+                exp.Alter, parse_one("ALTER TABLE `db`.`t` MODIFY COLUMN `a` BIGINT", read="doris")
             ),
-            parse_one(
-                "ALTER TABLE `db`.`t` MODIFY COLUMN `b` DATETIME(6)", into=exp.Alter, read="doris"
+            t.cast(
+                exp.Alter,
+                parse_one("ALTER TABLE `db`.`t` MODIFY COLUMN `b` DATETIME(6)", read="doris"),
             ),
         ]
     )
@@ -5064,6 +5065,7 @@ def test_iceberg_reading_time_travel_branch_and_compaction_ddl(
 
 def test_doris_4_1_search_indexes_compaction_and_sequence_col(
     make_mocked_engine_adapter: t.Callable[..., DorisEngineAdapter],
+    mocker: MockerFixture,
 ):
     adapter = make_mocked_engine_adapter(DorisEngineAdapter)
 
@@ -5172,3 +5174,29 @@ def test_doris_4_1_search_indexes_compaction_and_sequence_col(
     assert any(
         "PROPERTIES ('function_column.sequence_col'='updated_at')" in call for call in sql_calls
     )
+
+    # 5. Self-referencing replace_query with aliased, default-catalog-qualified table reference preserves alias
+    adapter._default_catalog = "internal"
+    adapter.cursor.execute.reset_mock()
+    mocker.patch("sqlmesh.core.engine_adapter.doris.random_id", return_value="abcd1234")
+    mocker.patch.object(
+        adapter,
+        "get_data_object",
+        return_value=DataObject(
+            catalog="internal", schema="db", name="tbl", type=DataObjectType.TABLE
+        ),
+    )
+    adapter.replace_query(
+        "db.tbl",
+        parse_one("SELECT t.id, t.val FROM internal.db.tbl AS t", read="doris"),
+        target_columns_to_types={
+            "id": exp.DataType.build("INT", dialect="doris"),
+            "val": exp.DataType.build("INT", dialect="doris"),
+        },
+    )
+    self_ref_calls = to_sql_calls(adapter)
+    assert any(
+        "CREATE TABLE IF NOT EXISTS `internal`.`db`.`__temp_tbl_abcd1234`" in c
+        for c in self_ref_calls
+    )
+    assert any("FROM `internal`.`db`.`__temp_tbl_abcd1234` AS `t`" in c for c in self_ref_calls)