feat: Add IBM Db2 engine adapter with CI/CD integration

Adds a complete IBM Db2 LUW engine adapter for SQLMesh, including
connection configuration, Docker-based CI, integration tests, and
engine-specific documentation.

## What is included

### Core adapter (sqlmesh/core/engine_adapter/db2.py)
- Db2EngineAdapter implementing all standard SQLMesh engine operations
- SYSCAT-based metadata queries (columns, tables, schemas, indexes, views)
- CTAS with mandatory WITH DATA clause
- MERGE INTO with TARGET/SOURCE alias substitution
- CREATE INDEX with SYSCAT.INDEXES existence check
- CREATE/DROP SCHEMA using SYSCAT.SCHEMATA
- DROP VIEW using SYSCAT.VIEWS existence check
- GRANT/REVOKE using SYSCAT.TABAUTH (not INFORMATION_SCHEMA)
- Truncate implemented as DELETE FROM (Db2 has no TRUNCATE)
- TIMESTAMPTZ stripped to TIMESTAMP (SQL0180N fix)
- normalize_identifiers before quoting (SQL0204N fix)
- catalog_support = SINGLE_CATALOG_ONLY (Db2 databases are isolated)
- SUPPORTED_DROP_CASCADE_OBJECT_KINDS = [] (SQL0104N fix)
- MAX_IDENTIFIER_LENGTH = 128
- set_current_catalog via CONNECT TO

### Connection config (sqlmesh/core/config/connection.py)
- Db2ConnectionConfig with host/port/database/username/password/db2_schema
- SSL options (ssl, ssl_cert, ssl_key, ssl_ca)
- get_catalog() returns .upper() to match Db2 UPPERCASE normalisation
- Excluded from FORBIDDEN_STATE_SYNC_ENGINES with explanation comment
  (Db2 rejects table names starting with underscore)

### Framework integration
- sqlmesh/core/engine_adapter/__init__.py: conditional import guarded
  by Python >= 3.10 AND find_spec('db2_sqlglot') — prevents import
  crash in environments without the db2 extra installed
- sqlmesh/utils/migration.py: db2 added to MAX_TEXT_INDEX_LENGTH
  (255) and blob_text_type() returns VARCHAR(32000)

## CI / infrastructure
- .github/workflows/pr.yaml: db2 added to engine-tests-docker matrix
- .github/scripts/install-prerequisites.sh: db2 installs libxml2-dev
- .github/scripts/wait-for-db.sh: db2_ready() readiness probe
- tests/.../docker/compose.db2.yaml: IBM Db2 Community Edition image
- Makefile: db2-test target with junitxml; db2 added to install-dev
- pyproject.toml: db2 optional extra + pytest marker

## Tests
- tests/core/engine_adapter/test_db2.py: 22 unit tests (all passing)
- tests/core/engine_adapter/integration/test_integration_db2.py:
  18 Db2-specific integration tests
- tests/core/engine_adapter/integration/__init__.py: SYSCAT comment
  queries, role-based grant infrastructure for Db2
- tests/core/engine_adapter/integration/config.yaml: inttest_db2
  gateway with DuckDB state_connection
- tests/core/engine_adapter/integration/test_integration.py:
  skip blocks with documented reasons for 20 tests; adaptations
  for uppercase identifiers, TIMESTAMPTZ, grants

## CI results (stable baseline on Docker Db2 Community Edition)
- 81 passed · 46 skipped · 0 failed

## Known limitations and skipped tests

### SCD Type 2 (4 tests skipped)
Db2 SQL preprocessor treats identifiers starting with '_' as
conditional compilation directives (SQL20521N reason 7). SQLMesh
generates _exists, _key0, _row_number, _t as aliases. Additionally
SCD staging uses CTAS which Db2 does not support natively.
Fix: override _scd_type_2() in Db2EngineAdapter to rename aliases.

### View comments (3 tests skipped)
Db2 has no COMMENT ON VIEW statement (SQL0104N).

### test_sushi (1 test skipped)
CREATE SCHEMA IF NOT EXISTS is not valid Db2 SQL (SQL0104N).
Fix: add create_sql() override to db2-sqlglot-dialect to strip
IF NOT EXISTS from CREATE SCHEMA.

### ctx.create_context() tests (6 tests skipped)
shared.py:346 uses a case-sensitive == comparison between
catalog_name (TESTDB, Db2 uppercase) and _default_catalog
(testdb, duckdb-dialect lowercase). This is a one-line upstream
fix in shared.py that cannot be made in this PR:
  catalog_name.upper() != (engine_adapter._default_catalog or '').upper()

## Documentation
- docs/integrations/engines/db2.md: connection options, state
  connection guidance, limitations, example config
- docs/integrations/overview.md: Db2 entry added
- docs/guides/connections.md: Db2 link added
- mkdocs.yml: db2.md nav entry added

Signed-off-by: IBM Db2 Eco System <Hdm-dev-persona-db2-eco-system@ibm.com>

Author: hdm-db2-eco-system

.github/scripts/install-prerequisites.sh

@@ -17,6 +17,8 @@ ENGINE_DEPENDENCIES=""
 
 if [ "$ENGINE" == "spark" ]; then
     ENGINE_DEPENDENCIES="default-jdk"
+elif [ "$ENGINE" == "db2" ]; then
+    ENGINE_DEPENDENCIES="libxml2-dev build-essential"
 elif [ "$ENGINE" == "fabric" ]; then
     echo "Installing Microsoft package repository"
 

.github/scripts/wait-for-db.sh

@@ -90,6 +90,20 @@ risingwave_ready() {
     probe_port 4566
 }
 
+db2_ready() {
+    probe_port 50001
+
+    echo "Waiting for Db2 to finish initialising (this can take 2-4 minutes)..."
+    while true; do
+        if docker exec db2 su - db2inst1 -c "db2 connect to TESTDB" > /dev/null 2>&1; then
+            echo "Db2 is accepting connections"
+            break
+        fi
+        echo "Db2 not yet ready; sleeping 15s..."
+        sleep 15
+    done
+}
+
 echo "Waiting for $ENGINE to be ready..."
 
 READINESS_FUNC="${ENGINE}_ready"

.github/workflows/pr.yaml

@@ -252,7 +252,7 @@ jobs:
       fail-fast: false
       matrix:
         engine:
-          [duckdb, postgres, mysql, mssql, trino, spark, clickhouse, risingwave, starrocks]
+          [duckdb, postgres, mysql, mssql, trino, spark, clickhouse, risingwave, starrocks, db2]
     env:
       PYTEST_XDIST_AUTO_NUM_WORKERS: 2
       SQLMESH__DISABLE_ANONYMIZED_ANALYTICS: '1'

Makefile

@@ -14,7 +14,7 @@ else
 endif
 
 install-dev:
-	$(PIP) install -e ".[dev,web,slack,dlt,lsp]" ./examples/custom_materializations
+	$(PIP) install -e ".[dev,web,slack,dlt,lsp,db2]" ./examples/custom_materializations
 
 install-doc:
 	$(PIP) install -r ./docs/requirements.txt
@@ -222,6 +222,9 @@ risingwave-test: engine-risingwave-up
 
 starrocks-test: engine-starrocks-up
 	pytest -n auto -m "starrocks" --reruns 3 --junitxml=test-results/junit-starrocks.xml
+	
+db2-test: engine-db2-up
+	pytest -n auto -m "db2" --reruns 3 --junitxml=test-results/junit-db2.xml
 
 #################
 # Cloud Engines #

docs/guides/connections.md

@@ -84,6 +84,7 @@ default_gateway: local_db
 * [BigQuery](../integrations/engines/bigquery.md)
 * [ClickHouse](../integrations/engines/clickhouse.md)
 * [Databricks](../integrations/engines/databricks.md)
+* [Db2](../integrations/engines/db2.md)
 * [DuckDB](../integrations/engines/duckdb.md)
 * [Fabric](../integrations/engines/fabric.md)
 * [MotherDuck](../integrations/engines/motherduck.md)

docs/integrations/engines/db2.md

@@ -0,0 +1,75 @@
+# Db2
+
+This page provides information about how to use SQLMesh with [IBM Db2](https://www.ibm.com/products/db2).
+
+!!! info
+    The Db2 engine adapter is a community contribution. Due to this, only limited community support is available.
+
+## Local/Built-in Scheduler
+
+**Engine Adapter Type**: `db2`
+
+### Installation
+
+```
+pip install "sqlmesh[db2]"
+```
+
+### Connection options
+
+| Option              | Description                                                                                    | Type   | Required |
+|---------------------|------------------------------------------------------------------------------------------------|:------:|:--------:|
+| `type`              | Engine type name - must be `db2`                                                               | string | Y        |
+| `host`              | The hostname of the Db2 server                                                                 | string | Y        |
+| `port`              | The port number of the Db2 server. Default: `50000`                                            | int    | N        |
+| `database`          | The name of the Db2 database to connect to                                                     | string | Y        |
+| `username`          | The username to use for authentication with the Db2 server                                     | string | Y        |
+| `password`          | The password to use for authentication with the Db2 server                                     | string | Y        |
+| `db2_schema`        | Sets `CURRENTSCHEMA` on the connection. Controls the default schema for unqualified references. Typically set to the same value as `username`. | string | Y        |
+| `ssl`               | Enable TLS/SSL encryption. Default: `false`                                                    | bool   | N        |
+| `connect_timeout`   | The number of seconds to wait for the connection to the server. Default: `30`                  | int    | N        |
+| `concurrent_tasks`  | Maximum number of tasks to run concurrently. Default: `4`                                      | int    | N        |
+
+## Important Notes
+
+**State connection:** Db2 is **not supported** as a SQLMesh `state_connection`. Use DuckDB (recommended) or another supported engine for SQLMesh state storage:
+
+```yaml linenums="1"
+gateways:
+  db2:
+    connection:
+      type: db2
+      host: localhost
+      port: 50000
+      database: TESTDB
+      username: db2inst1
+      password: your_password
+      db2_schema: db2inst1
+    state_connection:
+      type: duckdb
+      database: ./state/sqlmesh_state.db
+
+default_gateway: db2
+
+model_defaults:
+  dialect: db2
+```
+
+**Table naming:** Db2 rejects table names that start with an underscore (`_`). SQLMesh's default physical table naming convention can generate names beginning with `_`. To avoid this, set `physical_table_naming_convention` to `hash_md5` in your project config:
+
+```yaml
+physical_table_naming_convention: hash_md5
+```
+
+## Limitations
+
+- **Single catalog only**: Db2 operates in single-catalog mode; cross-catalog queries are not supported.
+- **No inline column comments**: Column-level comments cannot be set inline during table creation.
+- **No atomic table replacement**: Db2 does not support `CREATE OR REPLACE TABLE`, so full model refreshes are not atomic. There is a brief window during which the table may be empty or partially populated.
+- **Identifier length**: Maximum identifier length is 128 characters.
+- **No `SELECT ... FOR UPDATE`**: Db2 does not support `SELECT ... FOR UPDATE` in the same way as OLTP databases; SQLMesh removes this clause when executing queries.
+
+## Resources
+
+- [IBM Db2 Documentation](https://www.ibm.com/docs/en/db2)
+- [IBM Db2 SQL Reference](https://www.ibm.com/docs/en/db2/11.5?topic=db2-sql)

docs/integrations/overview.md

@@ -16,6 +16,7 @@ SQLMesh supports the following execution engines for running SQLMesh projects (e
 * [BigQuery](./engines/bigquery.md) (bigquery)
 * [ClickHouse](./engines/clickhouse.md) (clickhouse)
 * [Databricks](./engines/databricks.md) (databricks)
+* [Db2](./engines/db2.md) (db2)
 * [DuckDB](./engines/duckdb.md) (duckdb)
 * [Fabric](./engines/fabric.md) (fabric)
 * [MotherDuck](./engines/motherduck.md) (motherduck)

mkdocs.yml

@@ -82,6 +82,7 @@ nav:
       - integrations/engines/bigquery.md
       - integrations/engines/clickhouse.md
       - integrations/engines/databricks.md
+      - integrations/engines/db2.md
       - integrations/engines/duckdb.md
       - integrations/engines/fabric.md
       - integrations/engines/motherduck.md

pyproject.toml

@@ -112,6 +112,10 @@ dev = [
 ]
 dbt = ["dbt-core<2"]
 dlt = ["dlt"]
+db2 = [
+    "ibm_db",
+    "db2-sqlglot-dialect;python_version>=\"3.10\""
+]
 duckdb = []
 fabric = ["pyodbc>=5.0.0"]
 fabric-mssql-python = ["mssql-python>=1.1.0;python_version>=\"3.10\""]
@@ -270,6 +274,7 @@ markers = [
     "clickhouse: test for Clickhouse (standalone mode / cluster mode)",
     "clickhouse_cloud: test for Clickhouse (cloud mode)",
     "databricks: test for Databricks",
+    "db2: test for Db2",
     "duckdb: test for DuckDB",
     "fabric: test for Fabric",
     "motherduck: test for MotherDuck",

sqlmesh/core/config/connection.py

@@ -53,6 +53,9 @@
     "mssql",
     "azuresql",
 }
+# Note: Db2 is excluded because it doesn't allow table names starting with underscore (_)
+# which SQLMesh uses for state tables (_versions, _snapshots, _environments, _intervals).
+# Use a separate state_connection (e.g., DuckDB) for Db2 gateways.
 FORBIDDEN_STATE_SYNC_ENGINES = {
     # Do not support row-level operations
     "spark",
@@ -2602,6 +2605,91 @@ def _connection_factory(self) -> t.Callable:
     BaseDuckDBConnectionConfig,  # type: ignore[type-abstract]
 }
 
+
+class Db2ConnectionConfig(ConnectionConfig):
+    host: str
+    port: int = 50000
+    database: str
+    db2_schema: str
+    username: str
+    password: str
+    ssl: bool = False
+    ssl_cert: t.Optional[str] = None
+    ssl_key: t.Optional[str] = None
+    ssl_ca: t.Optional[str] = None
+    connect_timeout: int = 30
+
+    concurrent_tasks: int = 4
+    register_comments: bool = True
+    pre_ping: bool = True
+
+    type_: t.Literal["db2"] = Field(alias="type", default="db2")
+    DIALECT: t.ClassVar[t.Literal["db2"]] = "db2"
+    DISPLAY_NAME: t.ClassVar[t.Literal["Db2"]] = "Db2"
+    DISPLAY_ORDER: t.ClassVar[t.Literal[19]] = 19
+
+    _engine_import_validator = _get_engine_import_validator("ibm_db", "db2")
+
+    @property
+    def _connection_kwargs_keys(self) -> t.Set[str]:
+        return {
+            "host",
+            "port",
+            "database",
+            "db2_schema",
+            "username",
+            "password",
+        }
+
+    @property
+    def _engine_adapter(self) -> t.Type[EngineAdapter]:
+        # DB2 adapter requires Python 3.10+ for db2-sqlglot-dialect
+        # Use getattr to avoid mypy errors on Python 3.9
+        return t.cast(
+            t.Type[EngineAdapter], getattr(engine_adapter, "Db2EngineAdapter", EngineAdapter)
+        )
+
+    def get_catalog(self) -> t.Optional[str]:
+        """Db2 stores catalog names in uppercase; normalise here so the default_catalog
+        passed to the adapter matches what get_current_catalog() returns at runtime."""
+        catalog = super().get_catalog()
+        return catalog.upper() if catalog else None
+
+    @property
+    def _connection_factory(self) -> t.Callable:
+        import ibm_db_dbi  # type: ignore
+
+        ssl = self.ssl
+        ssl_cert = self.ssl_cert
+        ssl_key = self.ssl_key
+        ssl_ca = self.ssl_ca
+        connect_timeout = self.connect_timeout
+
+        def connect_db2(**kwargs: t.Any) -> t.Any:
+            conn_str_parts = [
+                f"DATABASE={kwargs['database']}",
+                f"HOSTNAME={kwargs['host']}",
+                f"PORT={kwargs['port']}",
+                "PROTOCOL=TCPIP",
+                f"UID={kwargs['username']}",
+                f"PWD={kwargs['password']}",
+                f"CURRENTSCHEMA={kwargs['db2_schema']}",
+                f"CONNECTTIMEOUT={connect_timeout}",
+            ]
+            if ssl:
+                conn_str_parts.append("SECURITY=SSL")
+                if ssl_cert:
+                    conn_str_parts.append(f"SSLClientCertificate={ssl_cert}")
+                if ssl_key:
+                    conn_str_parts.append(f"SSLClientKey={ssl_key}")
+                if ssl_ca:
+                    conn_str_parts.append(f"SSLServerCertificate={ssl_ca}")
+            conn_str = ";".join(conn_str_parts) + ";"
+            return ibm_db_dbi.connect(conn_str, "", "")
+
+        return connect_db2
+
+
 CONNECTION_CONFIG_TO_TYPE = {
     # Map all subclasses of ConnectionConfig to the value of their `type_` field.
     tpe.all_field_infos()["type_"].default: tpe