Repository: TobikoData/sqlmesh
Pull request #3140 (2024-09-16 22:26:21)

Fixed daemon detection to determine forking behavior
Introduced the `is_daemon_process` function in `utils/system.py` to determine if the current process is a daemon. Updated the logic in `constants.py` to use this new utility function for determining forking behavior.

CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/TobikoData/sqlmesh?pullRequest=3140) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://cla-assistant.io/TobikoData/sqlmesh?pullRequest=3140) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/TobikoData/sqlmesh?pullRequest=3140) it.</sub>
izeigerman: Hey @sammywachtel ! Thanks a lot for your submission. Can you please provide a little more context around the issue you ran into with the existing implementation and how the new approach is better.

Patch:
diff --git a/sqlmesh/core/constants.py b/sqlmesh/core/constants.py
index 5c742d5c..0346ab78 100644
--- a/sqlmesh/core/constants.py
+++ b/sqlmesh/core/constants.py
@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 import datetime
-import multiprocessing as mp
 import os
+import sys
 import typing as t
 from pathlib import Path
 
@@ -32,11 +32,27 @@ MAX_MODEL_DEFINITION_SIZE = 10000
 """Maximum number of characters in a model definition"""
 
 
+def is_daemon_process() -> bool:
+    """
+    Determines whether the current process is running as a daemon (background process).
+
+    This function checks if the standard output (stdout) is connected to a terminal.
+    It does this by calling `sys.stdout.fileno()` to retrieve the file descriptor
+    for the stdout stream and `os.isatty()` to check if this file descriptor is
+    associated with a terminal device.
+
+    Returns:
+        bool: True if the process is running as a daemon (not attached to a terminal),
+              False if the process is running in a terminal (interactive mode).
+    """
+    return not os.isatty(sys.stdout.fileno())
+
+
 # The maximum number of fork processes, used for loading projects
 # None means default to process pool, 1 means don't fork, :N is number of processes
 # Factors in the number of available CPUs even if the process is bound to a subset of them
 # (e.g. via taskset) to avoid oversubscribing the system and causing kill signals
-if hasattr(os, "fork") and not mp.current_process().daemon:
+if hasattr(os, "fork") and not is_daemon_process():
     try:
         MAX_FORK_WORKERS: t.Optional[int] = int(os.getenv("MAX_FORK_WORKERS"))  # type: ignore
     except TypeError:
@@ -81,7 +97,6 @@ SQLMESH_MACRO = "__sqlmesh__macro__"
 SQLMESH_BUILTIN = "__sqlmesh__builtin__"
 SQLMESH_METADATA = "__sqlmesh__metadata__"
 
-
 BUILTIN = "builtin"
 AIRFLOW = "airflow"
 DBT = "dbt"
diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
index 1a4f134d..e93be512 100644
--- a/sqlmesh/core/dialect.py
+++ b/sqlmesh/core/dialect.py
@@ -579,7 +579,12 @@ def _props_sql(self: Generator, expressions: t.List[exp.Expression]) -> str:
     size = len(expressions)
 
     for i, prop in enumerate(expressions):
-        sql = self.indent(f"{prop.name} {self.sql(prop, 'value')}")
+        value = prop.args.get("value")
+        if prop.name == "when_matched" and isinstance(value, list):
+            output_value = ", ".join(self.sql(v) for v in value)
+        else:
+            output_value = self.sql(prop, "value")
+        sql = self.indent(f"{prop.name} {output_value}")
 
         if i < size - 1:
             sql += ","


Test patch:
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 51217d1a..92534d82 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -3916,7 +3916,7 @@ def test_when_matched_multiple():
     expressions = d.parse(
         """
         MODEL (
-          name db.employees,
+          name @{schema}.employees,
           kind INCREMENTAL_BY_UNIQUE_KEY (
             unique_key name,
             when_matched WHEN MATCHED AND source.x = 1 THEN UPDATE SET target.salary = COALESCE(source.salary, target.salary),
@@ -3933,7 +3933,7 @@ def test_when_matched_multiple():
         "WHEN MATCHED THEN UPDATE SET __MERGE_TARGET__.salary = COALESCE(__MERGE_SOURCE__.salary, __MERGE_TARGET__.salary)",
     ]
 
-    model = load_sql_based_model(expressions, dialect="hive")
+    model = load_sql_based_model(expressions, dialect="hive", variables={"schema": "db"})
     assert len(model.kind.when_matched) == 2
     assert model.kind.when_matched[0].sql() == expected_when_matched[0]
     assert model.kind.when_matched[1].sql() == expected_when_matched[1]