Merge branch 'TobikoData:main' into doris

Author: xinge-ji

sqlmesh/core/renderer.py

@@ -30,7 +30,7 @@
     SQLMeshError,
     raise_config_error,
 )
-from sqlmesh.utils.jinja import JinjaMacroRegistry
+from sqlmesh.utils.jinja import JinjaMacroRegistry, extract_error_details
 from sqlmesh.utils.metaprogramming import Executable, prepare_env
 
 if t.TYPE_CHECKING:
@@ -239,18 +239,24 @@ def _resolve_table(table: str | exp.Table) -> str:
                 logger.debug(
                     f"Rendered Jinja expression for model '{self._model_fqn}' at '{self._path}': '{rendered_expression}'"
                 )
-                if rendered_expression.strip():
-                    expressions = [e for e in parse(rendered_expression, read=self._dialect) if e]
-
-                    if not expressions:
-                        raise ConfigError(f"Failed to parse an expression:\n{self._expression}")
             except ParsetimeAdapterCallError:
                 raise
             except Exception as ex:
                 raise ConfigError(
-                    f"Could not render or parse jinja at '{self._path}'.\n{ex}"
+                    f"Could not render jinja for '{self._path}'.\n" + extract_error_details(ex)
                 ) from ex
 
+            if rendered_expression.strip():
+                try:
+                    expressions = [e for e in parse(rendered_expression, read=self._dialect) if e]
+
+                    if not expressions:
+                        raise ConfigError(f"Failed to parse an expression:\n{self._expression}")
+                except Exception as ex:
+                    raise ConfigError(
+                        f"Could not parse the rendered jinja at '{self._path}'.\n{ex}"
+                    ) from ex
+
         if this_model:
             render_kwargs["this_model"] = this_model
 

sqlmesh/dbt/adapter.py

@@ -168,7 +168,8 @@ def compare_dbr_version(self, major: int, minor: int) -> int:
 
     @property
     def graph(self) -> t.Any:
-        return AttributeDict(
+        flat_graph = self.jinja_globals.get("flat_graph", None)
+        return flat_graph or AttributeDict(
             {
                 "exposures": {},
                 "groups": {},
@@ -276,10 +277,6 @@ def __init__(
             **table_mapping,
         }
 
-    @property
-    def graph(self) -> t.Any:
-        return self.jinja_globals.get("flat_graph", super().graph)
-
     def get_relation(
         self, database: t.Optional[str], schema: str, identifier: str
     ) -> t.Optional[BaseRelation]:

sqlmesh/dbt/builtin.py

@@ -502,8 +502,14 @@ def create_builtin_globals(
         else:
             builtin_globals["model"] = AttributeDict(model.copy())
 
+    builtin_globals["flags"] = (
+        Flags(which="run") if engine_adapter is not None else Flags(which="parse")
+    )
+    builtin_globals["invocation_args_dict"] = {
+        k.lower(): v for k, v in builtin_globals["flags"].__dict__.items()
+    }
+
     if engine_adapter is not None:
-        builtin_globals["flags"] = Flags(which="run")
         adapter: BaseAdapter = RuntimeAdapter(
             engine_adapter,
             jinja_macros,
@@ -521,7 +527,6 @@ def create_builtin_globals(
             project_dialect=project_dialect,
         )
     else:
-        builtin_globals["flags"] = Flags(which="parse")
         adapter = ParsetimeAdapter(
             jinja_macros,
             jinja_globals={**builtin_globals, **jinja_globals},

sqlmesh/utils/jinja.py

@@ -7,8 +7,11 @@
 import zlib
 from collections import defaultdict
 from enum import Enum
+from sys import exc_info
+from traceback import walk_tb
 
-from jinja2 import Environment, Template, nodes
+from jinja2 import Environment, Template, nodes, UndefinedError
+from jinja2.runtime import Macro
 from sqlglot import Dialect, Expression, Parser, TokenType
 
 from sqlmesh.core import constants as c
@@ -664,3 +667,24 @@ def make_jinja_registry(
     jinja_registry = jinja_registry.trim(jinja_references)
 
     return jinja_registry
+
+
+def extract_error_details(ex: Exception) -> str:
+    """Extracts a readable message from a Jinja2 error, to include missing name and macro."""
+
+    error_details = ""
+    if isinstance(ex, UndefinedError):
+        if match := re.search(r"'(\w+)'", str(ex)):
+            error_details += f"\nUndefined macro/variable: '{match.group(1)}'"
+        try:
+            _, _, exc_traceback = exc_info()
+            for frame, _ in walk_tb(exc_traceback):
+                if frame.f_code.co_name == "_invoke":
+                    macro = frame.f_locals.get("self")
+                    if isinstance(macro, Macro):
+                        error_details += f" in macro: '{macro.name}'\n"
+                        break
+        except:
+            # to fall back to the generic error message if frame analysis fails
+            pass
+    return error_details or str(ex)

tests/core/test_context.py

@@ -631,6 +631,49 @@ def test_env_and_default_schema_normalization(mocker: MockerFixture):
     assert list(context.fetchdf('select c from "DEFAULT__DEV"."X"')["c"])[0] == 1
 
 
+def test_jinja_macro_undefined_variable_error(tmp_path: pathlib.Path):
+    models_dir = tmp_path / "models"
+    models_dir.mkdir(parents=True)
+    macros_dir = tmp_path / "macros"
+    macros_dir.mkdir(parents=True)
+
+    macro_file = macros_dir / "my_macros.sql"
+    macro_file.write_text("""
+{%- macro generate_select(table_name) -%}
+  {%- if target.name == 'production' -%}
+    {%- set results = run_query('SELECT 1') -%}
+  {%- endif -%}
+  SELECT {{ results.columns[0].values()[0] }} FROM {{ table_name }}
+{%- endmacro -%}
+""")
+
+    model_file = models_dir / "my_model.sql"
+    model_file.write_text("""
+MODEL (
+    name my_schema.my_model,
+    kind FULL
+);
+
+JINJA_QUERY_BEGIN;
+{{ generate_select('users') }}
+JINJA_END;
+""")
+
+    config_file = tmp_path / "config.yaml"
+    config_file.write_text("""
+model_defaults:
+  dialect: duckdb
+""")
+
+    with pytest.raises(ConfigError) as exc_info:
+        Context(paths=str(tmp_path))
+
+    error_message = str(exc_info.value)
+    assert "Failed to load model" in error_message
+    assert "Could not render jinja for" in error_message
+    assert "Undefined macro/variable: 'target' in macro: 'generate_select'" in error_message
+
+
 def test_clear_caches(tmp_path: pathlib.Path):
     models_dir = tmp_path / "models"
 
@@ -2497,7 +2540,7 @@ def test_plan_min_intervals(tmp_path: Path):
       ),
       start '2020-01-01',
       cron '@daily'
-    );                        
+    );
 
     select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;
     """)
@@ -2510,9 +2553,9 @@ def test_plan_min_intervals(tmp_path: Path):
       ),
       start '2020-01-01',
       cron '@weekly'
-    );                        
+    );
 
-    select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;                        
+    select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;
     """)
 
     (tmp_path / "models" / "monthly_model.sql").write_text("""
@@ -2523,9 +2566,9 @@ def test_plan_min_intervals(tmp_path: Path):
       ),
       start '2020-01-01',
       cron '@monthly'
-    );                        
+    );
 
-    select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;                         
+    select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;
     """)
 
     (tmp_path / "models" / "ended_daily_model.sql").write_text("""
@@ -2537,9 +2580,9 @@ def test_plan_min_intervals(tmp_path: Path):
       start '2020-01-01',
       end '2020-01-18',
       cron '@daily'
-    );                        
+    );
 
-    select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;                 
+    select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;
     """)
 
     context.load()
@@ -2672,7 +2715,7 @@ def test_plan_min_intervals_adjusted_for_downstream(tmp_path: Path):
       ),
       start '2020-01-01',
       cron '@hourly'
-    );                        
+    );
 
     select @start_dt as start_dt, @end_dt as end_dt;
     """)
@@ -2681,11 +2724,11 @@ def test_plan_min_intervals_adjusted_for_downstream(tmp_path: Path):
     MODEL (
       name sqlmesh_example.two_hourly_model,
       kind INCREMENTAL_BY_TIME_RANGE (
-        time_column start_dt        
+        time_column start_dt
       ),
       start '2020-01-01',
       cron '0 */2 * * *'
-    );                        
+    );
 
     select start_dt, end_dt from sqlmesh_example.hourly_model where start_dt between @start_dt and @end_dt;
     """)
@@ -2694,11 +2737,11 @@ def test_plan_min_intervals_adjusted_for_downstream(tmp_path: Path):
     MODEL (
       name sqlmesh_example.unrelated_monthly_model,
       kind INCREMENTAL_BY_TIME_RANGE (
-        time_column start_dt        
+        time_column start_dt
       ),
       start '2020-01-01',
       cron '@monthly'
-    );                        
+    );
 
     select @start_dt as start_dt, @end_dt as end_dt;
     """)
@@ -2711,7 +2754,7 @@ def test_plan_min_intervals_adjusted_for_downstream(tmp_path: Path):
       ),
       start '2020-01-01',
       cron '@daily'
-    );                        
+    );
 
     select start_dt, end_dt from sqlmesh_example.hourly_model where start_dt between @start_dt and @end_dt;
     """)
@@ -2724,7 +2767,7 @@ def test_plan_min_intervals_adjusted_for_downstream(tmp_path: Path):
       ),
       start '2020-01-01',
       cron '@weekly'
-    );                        
+    );
 
     select start_dt, end_dt from sqlmesh_example.daily_model where start_dt between @start_dt and @end_dt;
     """)

tests/core/test_model.py

@@ -8588,6 +8588,23 @@ def test_comments_in_jinja_query():
         model.render_query()
 
 
+def test_jinja_render_parse_error():
+    expressions = d.parse(
+        """
+        MODEL (name db.test_model);
+
+        JINJA_QUERY_BEGIN;
+        {{ unknown_macro() }}
+        JINJA_END;
+        """
+    )
+
+    model = load_sql_based_model(expressions)
+
+    with pytest.raises(ConfigError, match=r"Could not render jinja"):
+        model.render_query()
+
+
 def test_jinja_render_debug_logging(caplog):
     """Test that rendered Jinja expressions are logged for debugging."""
     import logging
@@ -8609,7 +8626,7 @@ def test_jinja_render_debug_logging(caplog):
     model = load_sql_based_model(expressions)
 
     # Attempt to render - this should fail due to invalid SQL syntax
-    with pytest.raises(ConfigError, match=r"Could not render or parse jinja"):
+    with pytest.raises(ConfigError, match=r"Could not parse the rendered jinja"):
         model.render_query()
 
     # Check that the rendered Jinja was logged

tests/dbt/test_model.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 
 from sqlglot import exp
+from sqlglot.errors import SchemaError
 from sqlmesh import Context
 from sqlmesh.core.model import TimeColumn, IncrementalByTimeRangeKind
 from sqlmesh.core.model.kind import OnDestructiveChange, OnAdditiveChange
@@ -579,3 +580,56 @@ def test_load_microbatch_with_ref_no_filter(
         context.render(microbatch_two_snapshot_fqn, start="2025-01-01", end="2025-01-10").sql()
         == 'SELECT "microbatch"."cola" AS "cola", "microbatch"."ds" AS "ds" FROM "local"."main"."microbatch" AS "microbatch"'
     )
+
+
+@pytest.mark.slow
+def test_dbt_jinja_macro_undefined_variable_error(create_empty_project):
+    project_dir, model_dir = create_empty_project()
+
+    dbt_profile_config = {
+        "test": {
+            "outputs": {
+                "duckdb": {
+                    "type": "duckdb",
+                    "path": str(project_dir.parent / "dbt_data" / "main.db"),
+                }
+            },
+            "target": "duckdb",
+        }
+    }
+    db_profile_file = project_dir / "profiles.yml"
+    with open(db_profile_file, "w", encoding="utf-8") as f:
+        YAML().dump(dbt_profile_config, f)
+
+    macros_dir = project_dir / "macros"
+    macros_dir.mkdir()
+
+    # the execute guard in the macro is so that dbt won't fail on the manifest loading earlier
+    macro_file = macros_dir / "my_macro.sql"
+    macro_file.write_text("""
+{%- macro select_columns(table_name) -%}
+  {% if execute %}
+    {%- if target.name == 'production' -%}
+        {%- set columns = run_query('SELECT column_name FROM information_schema.columns WHERE table_name = \'' ~ table_name ~ '\'') -%}
+    {%- endif -%}
+    SELECT {{ columns.rows[0][0] }} FROM {{ table_name }}
+  {%- endif -%}
+{%- endmacro -%}
+""")
+
+    model_file = model_dir / "my_model.sql"
+    model_file.write_text("""
+{{ config(
+    materialized='table'
+) }}
+
+{{ select_columns('users') }}
+""")
+
+    with pytest.raises(SchemaError) as exc_info:
+        Context(paths=project_dir)
+
+    error_message = str(exc_info.value)
+    assert "Failed to update model schemas" in error_message
+    assert "Could not render jinja for" in error_message
+    assert "Undefined macro/variable: 'columns' in macro: 'select_columns'" in error_message