Repository: TobikoData/sqlmesh
Pull request #2049 (2024-01-30 14:54:50)

fix: remove pydantic 2.6 restriction
For dev we still need Pydantic <2.6 due to conflicts with Airflow (see comments) but users can use Pydantic 2.6+ if they want. We won't be testing 2.6+ in our CI/CD pipeline but I did run locally and verified all the slow tests passed. I'm hoping Airflow soon adds support for 2.6 and then we can update our dev install to use that latest Airflow release. 




Patch:
diff --git a/setup.py b/setup.py
index 7757e00e..48c1d371 100644
--- a/setup.py
+++ b/setup.py
@@ -43,7 +43,7 @@ setup(
         "ipywidgets",
         "jinja2",
         "pandas",
-        "pydantic<2.6.0",
+        "pydantic",
         "requests",
         "rich[jupyter]",
         "ruamel.yaml",
@@ -84,6 +84,11 @@ setup(
             "pandas-stubs",
             "psycopg2-binary",
             "pyarrow>=10.0.1,<10.1.0",
+            # All Airflow releases require flast-appbuilder==4.3.10 and
+            # 4.3.10 requires email-validator==1.3.1
+            # https://github.com/apache/airflow/blob/main/pyproject.toml#L685
+            # https://github.com/dpgaspar/Flask-AppBuilder/blob/master/requirements.txt#L25
+            "pydantic<2.6.0",
             "PyGithub",
             # 8.0.0 broke compatability with lazy-fixture
             "pytest<8.0.0",
diff --git a/sqlmesh/core/audit/definition.py b/sqlmesh/core/audit/definition.py
index 5232e21c..3af4b70e 100644
--- a/sqlmesh/core/audit/definition.py
+++ b/sqlmesh/core/audit/definition.py
@@ -485,8 +485,6 @@ def load_audit(
         raise
 
     meta_fields = {p.name: p.args.get("value") for p in meta.expressions if p}
-    if meta.comments:
-        meta_fields["description"] = "\n".join(comment.strip() for comment in meta.comments)
 
     standalone_field = meta_fields.pop("standalone", None)
     if standalone_field and not isinstance(standalone_field, exp.Boolean):
diff --git a/sqlmesh/core/notification_target.py b/sqlmesh/core/notification_target.py
index 21671430..531c40d6 100644
--- a/sqlmesh/core/notification_target.py
+++ b/sqlmesh/core/notification_target.py
@@ -217,7 +217,38 @@ class ConsoleNotificationTarget(BaseNotificationTarget):
             self.console.log_status_update(msg)
 
 
-class SlackWebhookNotificationTarget(BaseNotificationTarget):
+class BaseSlackNotificationTarget(BaseNotificationTarget):
+    def send(self, notification_status: NotificationStatus, msg: str, **kwargs: t.Any) -> None:
+        status_emoji = {
+            NotificationStatus.PROGRESS: slack.SlackAlertIcon.START,
+            NotificationStatus.SUCCESS: slack.SlackAlertIcon.SUCCESS,
+            NotificationStatus.FAILURE: slack.SlackAlertIcon.FAILURE,
+            NotificationStatus.WARNING: slack.SlackAlertIcon.WARNING,
+            NotificationStatus.INFO: slack.SlackAlertIcon.INFO,
+        }
+        composed = slack.message().add_primary_blocks(
+            slack.header_block(f"{status_emoji[notification_status]} SQLMesh Notification"),
+            slack.context_block(
+                f"*Status:* {notification_status.value}",
+                f"*Command:* {slack.stringify_list(sys.argv)}",
+            ),
+            slack.divider_block(),
+            slack.text_section_block(msg),
+            slack.context_block(f"*Python Version:* {sys.version}"),
+        )
+        self._send_slack_message(
+            composed=composed.slack_message,
+        )
+
+    def _send_slack_message(self, composed: slack.TSlackMessage) -> None:
+        """Send a composed message Slack.
+
+        Args:
+            composed: the formatted message to send to Slack
+        """
+
+
+class SlackWebhookNotificationTarget(BaseSlackNotificationTarget):
     url: t.Optional[str] = None
     type_: Literal["slack_webhook"] = Field(alias="type", default="slack_webhook")
     _client: t.Optional[WebhookClient] = None
@@ -238,27 +269,10 @@ class SlackWebhookNotificationTarget(BaseNotificationTarget):
             self._client = WebhookClient(url=self.url)
         return self._client
 
-    def send(self, notification_status: NotificationStatus, msg: str, **kwargs: t.Any) -> None:
-        status_emoji = {
-            NotificationStatus.PROGRESS: slack.SlackAlertIcon.START,
-            NotificationStatus.SUCCESS: slack.SlackAlertIcon.SUCCESS,
-            NotificationStatus.FAILURE: slack.SlackAlertIcon.FAILURE,
-            NotificationStatus.WARNING: slack.SlackAlertIcon.WARNING,
-            NotificationStatus.INFO: slack.SlackAlertIcon.INFO,
-        }
-        composed = slack.message().add_primary_blocks(
-            slack.header_block(f"{status_emoji[notification_status]} SQLMesh Notification"),
-            slack.context_block(
-                f"*Status:* {notification_status.value}",
-                f"*Command:* {slack.stringify_list(sys.argv)}",
-            ),
-            slack.divider_block(),
-            slack.text_section_block(msg),
-            slack.context_block(f"*Python Version:* {sys.version}"),
-        )
+    def _send_slack_message(self, composed: slack.TSlackMessage) -> None:
         self.client.send(
-            blocks=composed.slack_message["blocks"],
-            attachments=composed.slack_message["attachments"],  # type: ignore
+            blocks=composed["blocks"],
+            attachments=composed["attachments"],  # type: ignore
         )
 
     @property
@@ -266,7 +280,7 @@ class SlackWebhookNotificationTarget(BaseNotificationTarget):
         return bool(self.url)
 
 
-class SlackApiNotificationTarget(BaseNotificationTarget):
+class SlackApiNotificationTarget(BaseSlackNotificationTarget):
     token: t.Optional[str] = None
     channel: t.Optional[str] = None
     type_: Literal["slack_api"] = Field(alias="type", default="slack_api")
@@ -285,11 +299,15 @@ class SlackApiNotificationTarget(BaseNotificationTarget):
             self._client = WebClient(token=self.token)
         return self._client
 
-    def send(self, notification_status: NotificationStatus, msg: str, **kwargs: t.Any) -> None:
+    def _send_slack_message(self, composed: slack.TSlackMessage) -> None:
         if not self.channel:
             raise ConfigError("Missing Slack channel for notification")
 
-        self.client.chat_postMessage(channel=self.channel, text=msg)
+        self.client.chat_postMessage(
+            channel=self.channel,
+            blocks=composed["blocks"],
+            attachments=composed["attachments"],  # type: ignore
+        )
 
     @property
     def is_configured(self) -> bool:
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 494f5e8d..47dd87bb 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -232,14 +232,17 @@ class SnapshotEvaluator:
             deployability_index: Determines snapshots that are deployable in the context of this creation.
             on_complete: A callback to call on each successfully created snapshot.
         """
-        snapshots_with_table_names: t.List[t.Tuple[Snapshot, str]] = []
+        snapshots_with_table_names = defaultdict(set)
         tables_by_schema = defaultdict(set)
         for snapshot in target_snapshots:
             if not snapshot.is_model or snapshot.is_symbolic:
                 continue
-            table = exp.to_table(snapshot.table_name(False), dialect=snapshot.model.dialect)
-            snapshots_with_table_names.append((snapshot, table.name))
-            tables_by_schema[d.schema_(table.db, catalog=table.catalog)].add(table.name)
+            for is_deployable in (True, False):
+                table = exp.to_table(
+                    snapshot.table_name(is_deployable), dialect=snapshot.model.dialect
+                )
+                snapshots_with_table_names[snapshot].add(table.name)
+                tables_by_schema[d.schema_(table.db, catalog=table.catalog)].add(table.name)
 
         existing_objects: t.Set[str] = set()
         for schema, object_names in tables_by_schema.items():
@@ -248,8 +251,8 @@ class SnapshotEvaluator:
             existing_objects.update(obj.name for obj in objs)
 
         snapshots_to_create = []
-        for snapshot, table_name in snapshots_with_table_names:
-            if table_name not in existing_objects:
+        for snapshot, table_names in snapshots_with_table_names.items():
+            if table_names - existing_objects:
                 snapshots_to_create.append(snapshot)
             elif on_complete:
                 on_complete(snapshot)
diff --git a/web/client/src/library/components/graph/Graph.css b/web/client/src/library/components/graph/Graph.css
index 6599d76d..f4fd40ca 100644
--- a/web/client/src/library/components/graph/Graph.css
+++ b/web/client/src/library/components/graph/Graph.css
@@ -1,12 +1,17 @@
-.react-flow__node.react-flow__node-model:hover {
+.react-flow__node {
   z-index: 10 !important;
 }
-
+.react-flow__node.react-flow__node-model:active {
+  z-index: 20 !important;
+}
 react-flow__attribution {
   background: transparent;
 }
-
 .lineage__column-source b {
   font-weight: 900;
   color: var(--color-primary);
 }
+.react-flow__edge {
+  pointer-events: none !important;
+  z-index: -1 !important;
+}
diff --git a/web/client/src/library/components/graph/ModelLineage.tsx b/web/client/src/library/components/graph/ModelLineage.tsx
index 5abe15d1..104e86e0 100644
--- a/web/client/src/library/components/graph/ModelLineage.tsx
+++ b/web/client/src/library/components/graph/ModelLineage.tsx
@@ -427,7 +427,7 @@ function GraphControls({ nodes = [] }: { nodes: Node[] }): JSX.Element {
     <div className="px-2 flex items-center text-xs text-neutral-400 @container">
       <div className="contents">
         <Popover
-          className="flex @lg:hidden bg-none border-none py-1"
+          className="flex @lg:hidden bg-none border-none"
           aria-label="Show lineage node details"
         >
           <Popover.Button
diff --git a/web/client/src/library/components/graph/ModelLineageSearch.tsx b/web/client/src/library/components/graph/ModelLineageSearch.tsx
index 2cf64458..54c72bdf 100644
--- a/web/client/src/library/components/graph/ModelLineageSearch.tsx
+++ b/web/client/src/library/components/graph/ModelLineageSearch.tsx
@@ -38,7 +38,7 @@ export default function ModelLineageSearch({
       className={clsx(
         'w-full',
         showSearchInput
-          ? 'block absolute left-0 right-0 z-10 pr-10 bg-dark'
+          ? 'block absolute top-0 left-0 right-0 z-10 pr-10 bg-light dark:bg-dark @[40rem]:items-end @[40rem]:justify-end @[40rem]:flex @[40rem]:static @[40rem]:pr-0'
           : 'items-end justify-end flex',
       )}
     >


Test patch:
diff --git a/tests/core/engine_adapter/test_integration.py b/tests/core/engine_adapter/test_integration.py
index d39b51ea..f55ac11d 100644
--- a/tests/core/engine_adapter/test_integration.py
+++ b/tests/core/engine_adapter/test_integration.py
@@ -1376,6 +1376,24 @@ def test_sushi(ctx: TestContext):
     start = to_date(now() - timedelta(days=7))
     end = now()
 
+    # Databricks requires the table property `delta.columnMapping.mode = 'name'` for
+    # spaces in column names. Other engines error if it is set in the model definition,
+    # so we set it here.
+    if ctx.dialect == "databricks":
+        cust_rev_by_day_key = [key for key in context._models if "customer_revenue_by_day" in key][
+            0
+        ]
+
+        cust_rev_by_day_model_tbl_props = context._models[cust_rev_by_day_key].copy(
+            update={
+                "table_properties": {
+                    "delta.columnMapping.mode": exp.Literal(this="name", is_string=True)
+                }
+            }
+        )
+
+        context._models.update({cust_rev_by_day_key: cust_rev_by_day_model_tbl_props})
+
     context.plan(
         environment="test_prod",
         start=start,
diff --git a/tests/core/test_audit.py b/tests/core/test_audit.py
index 4fb103ae..6804f8e2 100644
--- a/tests/core/test_audit.py
+++ b/tests/core/test_audit.py
@@ -37,6 +37,7 @@ def model_default_catalog() -> Model:
 def test_load(assert_exp_eq):
     expressions = parse(
         """
+        -- Audit comment
         Audit (
             name my_audit,
             dialect spark,
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py
index 220b19a3..898c83fb 100644
--- a/tests/core/test_snapshot_evaluator.py
+++ b/tests/core/test_snapshot_evaluator.py
@@ -523,8 +523,7 @@ def test_evaluate_incremental_unmanaged(
         )
 
 
-def test_create_object_exists(mocker: MockerFixture, adapter_mock, make_snapshot):
-
+def test_create_tables_exists(mocker: MockerFixture, adapter_mock, make_snapshot):
     model = load_sql_based_model(
         parse(  # type: ignore
             """
@@ -547,13 +546,77 @@ def test_create_object_exists(mocker: MockerFixture, adapter_mock, make_snapshot
             schema="sqlmesh__test_schema",
             type=DataObjectType.VIEW,
         ),
+        DataObject(
+            name=f"test_schema__test_model__{snapshot.version}",
+            schema="sqlmesh__test_schema",
+            type=DataObjectType.VIEW,
+        ),
     ]
     evaluator = SnapshotEvaluator(adapter_mock)
 
     evaluator.create([snapshot], {})
     adapter_mock.create_view.assert_not_called()
     adapter_mock.get_data_objects.assert_called_once_with(
-        schema_("sqlmesh__test_schema"), {f"test_schema__test_model__{snapshot.version}__temp"}
+        schema_("sqlmesh__test_schema"),
+        {
+            f"test_schema__test_model__{snapshot.version}__temp",
+            f"test_schema__test_model__{snapshot.version}",
+        },
+    )
+
+
+def test_create_only_dev_table_exists(mocker: MockerFixture, adapter_mock, make_snapshot):
+    model = load_sql_based_model(
+        parse(  # type: ignore
+            """
+            MODEL (
+                name test_schema.test_model,
+                kind VIEW
+            );
+
+            SELECT a::int FROM tbl;
+            """
+        ),
+    )
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    adapter_mock.get_data_objects.return_value = [
+        DataObject(
+            name=f"test_schema__test_model__{snapshot.version}__temp",
+            schema="sqlmesh__test_schema",
+            type=DataObjectType.VIEW,
+        ),
+    ]
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    evaluator.create([snapshot], {})
+
+    common_kwargs = dict(
+        materialized=False,
+        table_properties={},
+        table_description=None,
+    )
+    rendered_query = model.render_query()
+    adapter_mock.create_view.assert_has_calls(
+        [
+            call(
+                snapshot.table_name(False),
+                rendered_query,
+                column_descriptions=None,
+                **common_kwargs,
+            ),
+            call(snapshot.table_name(), rendered_query, column_descriptions={}, **common_kwargs),
+        ]
+    )
+
+    adapter_mock.get_data_objects.assert_called_once_with(
+        schema_("sqlmesh__test_schema"),
+        {
+            f"test_schema__test_model__{snapshot.version}__temp",
+            f"test_schema__test_model__{snapshot.version}",
+        },
     )