fix(dlt): compare dlt version semantically for destination client

The dlt destination-client gate compared the version as a string
(`dlt.__version__ >= "1.10.0"`). String comparison is lexicographic, so
it wrongly evaluates "1.9.0" >= "1.10.0" (and "1.2.0", "1.5.10", ...) as
True. That sent every 1.x version down the >=1.10.0 branch and made the
pre-1.10 `_sql_job_client` fallback unreachable.

Use `packaging.version.parse` for a semantic comparison, mirroring the
existing idiom in sqlmesh/core/config/connection.py. The client selection
is extracted into a small `_get_destination_client` helper so the version
gate can be unit tested without a live dlt pipeline.

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

Author: anxkhn

sqlmesh/integrations/dlt.py

@@ -1,13 +1,29 @@
 import typing as t
 import click
 from datetime import datetime, timedelta, timezone
+from packaging import version
 from pydantic import ValidationError
 from sqlglot import exp, parse_one
 from sqlmesh.core.config.connection import parse_connection_config
 from sqlmesh.core.context import Context
 from sqlmesh.utils.date import yesterday_ds
 
 
+def _get_destination_client(pipeline: t.Any, schema: t.Any) -> t.Any:
+    """Return the dlt destination client, handling the 1.10.0 API change.
+
+    dlt 1.10.0 replaced the private ``_sql_job_client`` with the public
+    ``destination_client``. The version must be compared semantically rather
+    than lexicographically: a string comparison wrongly evaluates
+    ``"1.9.0" >= "1.10.0"`` as ``True``.
+    """
+    import dlt
+
+    if version.parse(dlt.__version__) >= version.parse("1.10.0"):
+        return pipeline.destination_client()
+    return pipeline._sql_job_client(schema)  # type: ignore
+
+
 def generate_dlt_models_and_settings(
     pipeline_name: str,
     dialect: str,
@@ -63,10 +79,7 @@ def generate_dlt_models_and_settings(
     if db_type == "filesystem":
         connection_config = None
     else:
-        if dlt.__version__ >= "1.10.0":
-            client = pipeline.destination_client()
-        else:
-            client = pipeline._sql_job_client(schema)  # type: ignore
+        client = _get_destination_client(pipeline, schema)
         config = client.config
         credentials = config.credentials
         configs = {

tests/integrations/test_dlt.py

@@ -1,4 +1,53 @@
-from sqlmesh.integrations.dlt import generate_incremental_model
+import typing as t
+
+import pytest
+
+from sqlmesh.integrations.dlt import _get_destination_client, generate_incremental_model
+
+
+class _FakePipeline:
+    """Records which destination-client accessor the version gate selects."""
+
+    def __init__(self) -> None:
+        self.called: t.Optional[str] = None
+
+    def destination_client(self) -> str:
+        self.called = "destination_client"
+        return self.called
+
+    def _sql_job_client(self, schema: t.Any) -> str:
+        self.called = "_sql_job_client"
+        return self.called
+
+
+@pytest.mark.parametrize(
+    "dlt_version, expected",
+    [
+        ("1.9.0", "_sql_job_client"),
+        ("1.2.0", "_sql_job_client"),
+        ("1.5.10", "_sql_job_client"),
+        ("1.10.0", "destination_client"),
+        ("1.28.1", "destination_client"),
+        ("2.0.0", "destination_client"),
+    ],
+)
+def test_get_destination_client_version_gate(
+    monkeypatch: pytest.MonkeyPatch, dlt_version: str, expected: str
+) -> None:
+    # The dlt destination-client API changed in 1.10.0: the public
+    # destination_client() replaced the private _sql_job_client(). The version
+    # gate must compare semantically (packaging.version.parse), not
+    # lexicographically. A string compare wrongly treats "1.9.0" >= "1.10.0" as
+    # True, which would send every 1.x version down the >=1.10.0 branch and make
+    # the pre-1.10 fallback unreachable.
+    import dlt
+
+    monkeypatch.setattr(dlt, "__version__", dlt_version)
+    pipeline = _FakePipeline()
+
+    _get_destination_client(pipeline, schema=None)
+
+    assert pipeline.called == expected
 
 
 def test_generate_incremental_model_filters_on_timestamp_macros() -> None: