feat(doris): support writable Paimon catalogs, Hive storage_format, and lazy default_catalog lookup

Author: brucearctor

docs/integrations/engines/doris.md

@@ -440,9 +440,34 @@ SQLMesh's Doris dialect natively parses and transpiles Apache Iceberg snapshot t
   );
   ```
 
-### Pattern 4: Reading External Lance Tables & Catalogs (Doris 4.1+)
+### Pattern 4: Reading & Writing External Apache Paimon and Apache Hive Catalogs
 
-Apache Doris 4.1+ supports reading multimodal and vector datasets stored in **Lance** format via external Lance catalogs (`'type' = 'lance'`) or table-valued functions (`'format' = 'lance'`).
+In addition to Apache Iceberg, `DorisEngineAdapter` supports reading and writing external **Apache Paimon** (`'type' = 'paimon'`, `table_format paimon`) and **Apache Hive** (`'type' = 'hms'`, `table_format hive`) catalogs:
+
+- **Apache Paimon (`table_format paimon`)**: Emits `ENGINE=paimon`, maps `storage_format` to `'file.format'` (`'parquet'` or `'orc'`), and automatically populates Paimon's `'primary-key'` property from `unique_key` (including partition columns when `partitioned_by` is set) so Doris 4.1+ row-level `DELETE` and `MERGE INTO` work out of the box.
+- **Apache Hive (`table_format hive`)**: Emits `ENGINE=hive` and maps `storage_format` to `'file_format'` (`'parquet'`, `'orc'`, or `'text'`). Supports `FULL` (`INSERT OVERWRITE TABLE`) and append workloads (`INCREMENTAL_UNMANAGED`), while row-level `DELETE` and `MERGE INTO` are not supported by Hive catalogs in Doris.
+
+```sql
+MODEL (
+  name paimon_catalog.silver.orders_paimon,
+  kind INCREMENTAL_BY_UNIQUE_KEY (
+    unique_key order_id
+  ),
+  table_format paimon,
+  storage_format parquet,
+  partitioned_by (ds)
+);
+
+SELECT
+  order_id::BIGINT AS order_id,
+  amount::DECIMAL(18, 2) AS amount,
+  ds::DATE AS ds
+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.
 
 ```yaml
 # config.yaml
@@ -472,8 +497,8 @@ SELECT
 FROM lance_catalog.multimodal.clip_embeddings;
 ```
 
-!!! warning "Lance Tables are Read-Only in Doris 4.1"
-    Doris 4.1 supports Lance catalogs and `format = 'lance'` table-valued functions for reading only (`CREATE TABLE`, `INSERT`, and `CTAS` are not supported by Doris on Lance catalogs). If a SQLMesh model is configured to materialize into a Lance catalog or sets `table_format = 'lance'`, `DorisEngineAdapter` raises a descriptive `SQLMeshError` directing you to materialize into the `internal` OLAP catalog or an `iceberg` catalog instead.
+!!! warning "Lance, Hudi, and Delta Lake Tables are Read-Only in Doris"
+    Doris supports Lance, Hudi, Delta Lake, and LakeSoul catalogs and `format = 'lance'` table-valued functions for reading only (`CREATE TABLE`, `INSERT`, and `CTAS` are not supported by Doris on those formats). If a SQLMesh model sets `table_format` (or `engine`) to `lance`, `hudi`, `deltalake`, or `lakesoul`, `DorisEngineAdapter` raises a descriptive `SQLMeshError` directing you to materialize into the `internal` OLAP catalog or a writable external catalog (`iceberg`, `paimon`, or `hive`).
 
 ---
 
@@ -746,18 +771,18 @@ MODEL (
 
 ## Supported Model Kinds & Catalog Matrix
 
-| SQLMesh Model Kind | Doris Internal OLAP Catalog (`internal` on Local Disk or S3/GCS Storage Vault) | External Apache Iceberg Catalog (`table_format iceberg` on S3/GCS/HDFS) | External Read-Only Lakehouse Catalogs (`lance`, `paimon`, `hudi`, `deltalake`) |
-| --- | :---: | :---: | :---: |
-| Source / Upstream `SELECT` | Supported | Supported (incl. `FOR TIME/VERSION AS OF`, `@branch`, `@tag`) | Supported (Read-Only via Catalog or S3/GCS TVF) |
-| `FULL` | Supported (`CREATE TABLE ... AS SELECT` / `DELETE` + `INSERT`) | Supported (`CREATE TABLE ... ENGINE=iceberg` / `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) | 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) | Read-Only in Doris (Write Guarded) |
-| `INCREMENTAL_BY_PARTITION` | Supported (Column & time-truncation range `DELETE` + `INSERT`) | Supported (`INSERT OVERWRITE` on Doris 2.1+; partition `DELETE` + `INSERT` requires Doris 4.1.0+ & Iceberg v2) | Read-Only in Doris (Write Guarded) |
-| `SCD_TYPE_2_BY_TIME` / `SCD_TYPE_2_BY_COLUMN` | Supported | Supported (Requires Doris 4.1.0+ & Iceberg v2 for row-level `DELETE`) | Read-Only in Doris (Write Guarded) |
-| `VIEW` | Supported (`CREATE OR REPLACE VIEW`) | Requires `internal` catalog (`environment_catalog_mapping`) | Requires `internal` catalog (`environment_catalog_mapping`) |
-| `VIEW (materialized true)` | Supported (Async MTMV & Sync MV with `materialized_type = 'SYNC'`) | Not supported by Doris inside external catalogs | Not supported by Doris inside external catalogs |
-| `SEED` / Python DataFrame Models | Supported (Automatic `VARCHAR(255)` OLAP key coercion) | Supported | Read-Only in Doris (Write Guarded) |
-| `CUSTOM ('doris_ddl_only')` | Supported (DDL + `ALTER TABLE` schema evolution only) | Supported | Not supported (Read-Only) |
+| 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) |
+| `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) |
+| `INCREMENTAL_BY_PARTITION` | Supported (Column & time-truncation range `DELETE` + `INSERT`) | Supported (Partition `DELETE` + `INSERT`; requires Doris 4.1.0+ & Iceberg v2 / Paimon PK) | Not supported by Hive in Doris (No `DELETE`) | Read-Only in Doris (Write Guarded) |
+| `SCD_TYPE_2_BY_TIME` / `SCD_TYPE_2_BY_COLUMN` | Supported | Supported (Requires Doris 4.1.0+ & Iceberg v2 / Paimon PK for row-level `DELETE`) | Not supported by Hive in Doris (No `DELETE`) | Read-Only in Doris (Write Guarded) |
+| `VIEW` | Supported (`CREATE OR REPLACE VIEW`) | Requires `internal` catalog (`environment_catalog_mapping`) | Requires `internal` catalog (`environment_catalog_mapping`) | Requires `internal` catalog (`environment_catalog_mapping`) |
+| `VIEW (materialized true)` | Supported (Async MTMV & Sync MV with `materialized_type = 'SYNC'`) | Not supported by Doris inside external catalogs | Not supported by Doris inside external catalogs | Not supported by Doris inside external catalogs |
+| `SEED` / Python DataFrame Models | Supported (Automatic `VARCHAR(255)` OLAP key coercion) | Supported | Supported | Read-Only in Doris (Write Guarded) |
+| `CUSTOM ('doris_ddl_only')` | Supported (DDL + `ALTER TABLE` schema evolution only) | Supported | Supported | Not supported (Read-Only) |
 
 ---
 

sqlmesh/core/engine_adapter/base.py

@@ -526,23 +526,37 @@ def replace_query(
             target_columns_to_types = self.columns(target_table)
         query = source_queries[0].query_factory()
 
-        default_catalog = self.default_catalog
+        quoted_target = quote_identifiers(target_table.copy())
+        default_catalog: t.Optional[str] = None
+        default_catalog_loaded = False
+        qualified_target: t.Optional[exp.Table] = None
 
-        def _qualified(table: exp.Table) -> exp.Table:
+        def _qualified(table: exp.Table, catalog: str) -> exp.Table:
             table = table.copy()
-            if default_catalog and not table.catalog and table.db:
-                table.set("catalog", exp.parse_identifier(default_catalog))
+            if not table.catalog and table.db:
+                table.set("catalog", exp.parse_identifier(catalog))
             return quote_identifiers(table)
 
-        quoted_target = quote_identifiers(target_table.copy())
-        qualified_target = _qualified(target_table)
-
         def _is_target_table(node: exp.Expr) -> bool:
+            nonlocal default_catalog, default_catalog_loaded, qualified_target
             if not isinstance(node, exp.Table):
                 return False
-            if quote_identifiers(node.copy()) == quoted_target:
+            quoted_node = quote_identifiers(node.copy())
+            if quoted_node == quoted_target:
                 return True
-            return bool(default_catalog) and _qualified(node) == qualified_target
+            if quoted_node.this != quoted_target.this or quoted_node.args.get(
+                "db"
+            ) != quoted_target.args.get("db"):
+                return False
+            if not default_catalog_loaded:
+                default_catalog = self.default_catalog
+                default_catalog_loaded = True
+                if default_catalog:
+                    qualified_target = _qualified(target_table, default_catalog)
+            return (
+                default_catalog is not None
+                and _qualified(node, default_catalog) == qualified_target
+            )
 
         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.

sqlmesh/core/engine_adapter/doris.py

@@ -353,7 +353,6 @@ def _restore_active_catalog_on_reconnect(self, sql: t.Optional[str] = None) -> N
 
     _READ_ONLY_EXTERNAL_FORMATS: t.ClassVar[t.Dict[str, str]] = {
         "lance": "Lance",
-        "paimon": "Paimon",
         "hudi": "Hudi",
         "deltalake": "Delta Lake",
         "delta": "Delta Lake",
@@ -412,11 +411,11 @@ def _is_iceberg_table(
     ) -> bool:
         if self._is_lance_table(table_format=table_format, table_properties=table_properties):
             return False
-        if table_format and str(table_format).strip().lower() in ("iceberg", "hive"):
+        if table_format and str(table_format).strip().lower() in ("iceberg", "paimon", "hive"):
             return True
         if table_properties:
             engine_val = self._extract_str_value(table_properties.get("engine"))
-            if engine_val and engine_val.strip().lower() in ("iceberg", "hive"):
+            if engine_val and engine_val.strip().lower() in ("iceberg", "paimon", "hive"):
                 return True
         effective_catalog = catalog_name or self.default_catalog
         if effective_catalog and str(effective_catalog).strip().lower() != "internal":
@@ -5454,23 +5453,28 @@ def _build_table_properties_exp(
             table_properties=table_properties_copy,
         )
 
-        # Pop Iceberg/Hive engine property if present so it is not duplicated in PROPERTIES (...)
+        # Pop Iceberg/Paimon/Hive engine property if present so it is not duplicated in PROPERTIES (...)
         explicit_engine = self._extract_str_value(table_properties_copy.pop("engine", None)) or (
             str(table_format).strip().lower() if table_format else None
         )
+        engine_lower = explicit_engine.strip().lower() if explicit_engine else None
         if is_iceberg and is_schema_ddl:
-            engine_name = (
-                "hive"
-                if explicit_engine and explicit_engine.strip().lower() == "hive"
-                else "iceberg"
-            )
+            engine_name = engine_lower if engine_lower in ("hive", "paimon") else "iceberg"
             properties.append(exp.EngineProperty(this=exp.var(engine_name)))
 
         if storage_format and not is_materialized_view:
             sf_str = str(storage_format).strip().strip("'").strip('"')
             if sf_str:
                 if is_iceberg:
-                    if (
+                    if engine_lower == "paimon":
+                        table_properties_copy.setdefault(
+                            "file.format", exp.Literal.string(sf_str.lower())
+                        )
+                    elif engine_lower == "hive":
+                        table_properties_copy.setdefault(
+                            "file_format", exp.Literal.string(sf_str.lower())
+                        )
+                    elif (
                         "write-format" not in table_properties_copy
                         and "write.format.default" not in table_properties_copy
                     ):
@@ -5613,6 +5617,24 @@ def _build_table_properties_exp(
                     chosen_part_by if isinstance(chosen_part_by, list) else [chosen_part_by]
                 )
 
+        if (
+            is_iceberg
+            and engine_lower == "paimon"
+            and unique_key_cols
+            and "primary-key" not in table_properties_copy
+        ):
+            paimon_pk_cols = [col_by_lower.get(k.lower(), k) for k in unique_key_cols]
+            if partitioned_by:
+                parsed_paimon_part, _ = self._parse_partition_expressions(partitioned_by)
+                pk_lower_set = {k.lower() for k in paimon_pk_cols}
+                for expr in parsed_paimon_part:
+                    for col_node in expr.find_all(exp.Column):
+                        part_col_name = col_by_lower.get(col_node.name.lower(), col_node.name)
+                        if part_col_name.lower() not in pk_lower_set:
+                            paimon_pk_cols.append(part_col_name)
+                            pk_lower_set.add(part_col_name.lower())
+            table_properties_copy["primary-key"] = exp.Literal.string(",".join(paimon_pk_cols))
+
         if not is_iceberg:
             specified_key_types = [
                 name

tests/core/engine_adapter/integration/test_integration_doris.py

@@ -105,6 +105,47 @@ def test_doris_iceberg_local_catalog(ctx: TestContext):
             read="doris",
         )
         adapter.execute(compact_expr, quote_identifiers=False)
+
+        # Also verify writable Apache Paimon local catalog DDL & schema evolution on Doris 4.1+
+        paimon_cat = f"local_paimon_{ctx.test_id}"
+        paimon_db = f"paimon_db_{ctx.test_id}"
+        paimon_tbl = f"{paimon_cat}.{paimon_db}.orders"
+        try:
+            adapter.execute(
+                f"""
+                CREATE CATALOG IF NOT EXISTS `{paimon_cat}` PROPERTIES (
+                    'type' = 'paimon',
+                    'warehouse' = 'file:///tmp/doris_paimon_warehouse/{paimon_cat}'
+                )
+                """,
+                quote_identifiers=False,
+            )
+            adapter.create_schema(f"{paimon_cat}.{paimon_db}")
+            adapter.create_table(
+                paimon_tbl,
+                {
+                    "id": exp.DataType.build("int"),
+                    "ds": exp.DataType.build("date"),
+                },
+                table_format="paimon",
+                storage_format="parquet",
+                partitioned_by=[exp.to_column("ds")],
+                unique_key=[exp.to_column("id")],
+            )
+            paimon_cols = adapter.columns(paimon_tbl)
+            assert "id" in paimon_cols
+            assert "ds" in paimon_cols
+        finally:
+            try:
+                adapter.drop_schema(
+                    f"{paimon_cat}.{paimon_db}", ignore_if_not_exists=True, cascade=True
+                )
+            except Exception:
+                pass
+            try:
+                adapter.execute(f"DROP CATALOG IF EXISTS `{paimon_cat}`", quote_identifiers=False)
+            except Exception:
+                pass
     finally:
         try:
             adapter.drop_schema(

tests/core/engine_adapter/test_doris.py

@@ -4875,14 +4875,32 @@ def test_lance_table_reading_and_write_guard(
             table_properties={"engine": exp.Literal.string("lance")},
         )
 
-    with pytest.raises(SQLMeshError, match="Doris Paimon tables are read-only"):
+    with pytest.raises(SQLMeshError, match="Doris Hudi tables are read-only"):
         adapter.create_table(
-            "paimon_cat.ml_db.out_tbl",
+            "hudi_cat.ml_db.out_tbl",
             target_columns_to_types={"id": exp.DataType.build("INT", dialect="doris")},
-            table_format="paimon",
+            table_format="hudi",
         )
 
-    # 4. External Hive catalog table creation emits ENGINE=hive and skips OLAP key reordering
+    # 4. External Paimon catalog table creation emits ENGINE=paimon, 'file.format', and 'primary-key'
+    adapter.cursor.execute.reset_mock()
+    adapter.create_table(
+        "paimon_cat.silver.orders",
+        target_columns_to_types={
+            "payload": exp.DataType.build("STRING", dialect="doris"),
+            "id": exp.DataType.build("INT", dialect="doris"),
+            "ds": exp.DataType.build("DATE", dialect="doris"),
+        },
+        table_format="paimon",
+        storage_format="parquet",
+        partitioned_by=[exp.to_column("ds")],
+        unique_key=[exp.to_column("id")],
+    )
+    assert to_sql_calls(adapter) == [
+        "CREATE TABLE IF NOT EXISTS `paimon_cat`.`silver`.`orders` (`payload` STRING, `id` INT, `ds` DATE) ENGINE=paimon PARTITION BY LIST (`ds`) () PROPERTIES ('file.format'='parquet', 'primary-key'='id,ds')"
+    ]
+
+    # 5. External Hive catalog table creation emits ENGINE=hive, 'file_format', and skips OLAP key reordering
     adapter.cursor.execute.reset_mock()
     adapter.create_table(
         "hive_cat.raw_db.events",
@@ -4891,9 +4909,10 @@ def test_lance_table_reading_and_write_guard(
             "id": exp.DataType.build("INT", dialect="doris"),
         },
         table_format="hive",
+        storage_format="orc",
     )
     assert to_sql_calls(adapter) == [
-        "CREATE TABLE IF NOT EXISTS `hive_cat`.`raw_db`.`events` (`payload` STRING, `id` INT) ENGINE=hive"
+        "CREATE TABLE IF NOT EXISTS `hive_cat`.`raw_db`.`events` (`payload` STRING, `id` INT) ENGINE=hive PROPERTIES ('file_format'='orc')"
     ]