feat(ingestion/lookml): resolve access notation for LookML Constant

metadata-ingestion/docs/sources/looker/looker_recipe.yml

@@ -8,4 +8,19 @@ source:
     client_id: ${LOOKER_CLIENT_ID}
     client_secret: ${LOOKER_CLIENT_SECRET}
 
+    # Liquid variables
+    # liquid_variables:
+    #   _user_attributes:
+    #     looker_env: "dev"
+    #     dev_database_prefix: "employee"
+    #     dev_schema_prefix: "public"
+    #   dw_eff_dt_date:
+    #     _is_selected: true
+    #   source_region: "ap-south-1"
+    #   db: "test-db"
+
+    # LookML Constant
+    # lookml_constants:
+    #   star_award_winner_year: "public.winner_2025"
+
 # sink configs

metadata-ingestion/docs/sources/looker/lookml_post.md

@@ -1,11 +1,27 @@
 #### Configuration Notes
 
-1. If a view contains a liquid template (e.g. `sql_table_name: {{ user_attributes['db']}}.kafka_streaming.events }}`, with `db=ANALYTICS_PROD`), then you will need to specify the values of those variables in the `liquid_variable` config as shown below:
-    ```yml
-      liquid_variable:
-        user_attributes:
-          db: ANALYTICS_PROD
-    ```
+1. If a view contains a liquid template (e.g. `sql_table_name: {{ user_attributes['db']}}.kafka_streaming.events }}`, with `db=ANALYTICS_PROD`), then you will need to specify the values of those variables in the `liquid_variables` config as shown below:
+
+   ```yml
+   liquid_variables:
+     user_attributes:
+       db: ANALYTICS_PROD
+   ```
+
+2. If your LookML code references a constant (e.g., `sql_table_name: @{db}.kafka_streaming.events;`), its value is resolved in the following order:
+
+- **First, checks the `manifest.lkml` file** for the constant definition.
+  ```manifest.lkml
+    constant: db {
+        value: "ANALYTICS_PROD"
+    }
+  ```
+- **If not found, falls back to `config`**
+
+  ```yml
+  lookml_constants:
+    db: ANALYTICS_PROD
+  ```
 
 ### Multi-Project LookML (Advanced)
 

metadata-ingestion/src/datahub/ingestion/source/looker/looker_dataclasses.py

@@ -75,6 +75,7 @@ def from_looker_dict(
             try:
                 parsed = load_and_preprocess_file(
                     path=included_file,
+                    reporter=reporter,
                     source_config=source_config,
                 )
                 included_explores = parsed.get("explores", [])
@@ -217,6 +218,7 @@ def resolve_includes(
                 try:
                     parsed = load_and_preprocess_file(
                         path=included_file,
+                        reporter=reporter,
                         source_config=source_config,
                     )
                     seen_so_far.add(included_file)

metadata-ingestion/src/datahub/ingestion/source/looker/looker_file_loader.py

@@ -1,7 +1,7 @@
 import logging
 import pathlib
 from dataclasses import replace
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 
 from datahub.ingestion.source.looker.looker_config import LookerConnectionDefinition
 from datahub.ingestion.source.looker.looker_dataclasses import LookerViewFile
@@ -30,12 +30,14 @@ def __init__(
         base_projects_folder: Dict[str, pathlib.Path],
         reporter: LookMLSourceReport,
         source_config: LookMLSourceConfig,
+        manifest_lookml_constant: Optional[List[Dict[str, str]]] = [],
     ) -> None:
         self.viewfile_cache: Dict[str, Optional[LookerViewFile]] = {}
         self._root_project_name = root_project_name
         self._base_projects_folder = base_projects_folder
         self.reporter = reporter
         self.source_config = source_config
+        self.manifest_lookml_constant = manifest_lookml_constant
 
     def _load_viewfile(
         self, project_name: str, path: str, reporter: LookMLSourceReport
@@ -73,7 +75,9 @@ def _load_viewfile(
 
             parsed = load_and_preprocess_file(
                 path=path,
+                reporter=self.reporter,
                 source_config=self.source_config,
+                manifest_lookml_constant=self.manifest_lookml_constant,
             )
 
             looker_viewfile = LookerViewFile.from_looker_dict(

metadata-ingestion/src/datahub/ingestion/source/looker/looker_template_language.py

@@ -27,6 +27,7 @@
 from datahub.ingestion.source.looker.lookml_config import (
     DERIVED_VIEW_PATTERN,
     LookMLSourceConfig,
+    LookMLSourceReport,
 )
 
 logger = logging.getLogger(__name__)
@@ -82,7 +83,9 @@
         return self._create_new_liquid_variables_with_default(variables=variables)
 
 
-def resolve_liquid_variable(text: str, liquid_variable: Dict[Any, Any]) -> str:
+def resolve_liquid_variable(
+    text: str, liquid_variable: Dict[Any, Any], report: LookMLSourceReport
+) -> str:
     # Set variable value to NULL if not present in liquid_variable dictionary
     Undefined.__str__ = lambda instance: "NULL"  # type: ignore
     try:
@@ -96,15 +99,22 @@
         # Resolve liquid template
         return create_template(text).render(liquid_variable)
     except LiquidSyntaxError as e:
-        logger.warning(f"Unsupported liquid template encountered. error [{e.message}]")
+        report.report_warning(
+            message="Unsupported liquid template",
+            context=f"text {text} liquid_variable {liquid_variable}",
+            exc=e,
+        )
         # TODO: There are some tag specific to looker and python-liquid library does not understand them. currently
         #  we are not parsing such liquid template.
         #
         # See doc: https://cloud.google.com/looker/docs/templated-filters and look for { % condition region %}
         # order.region { % endcondition %}
     except CustomTagException as e:
-        logger.warning(e)
-        logger.debug(e, exc_info=e)
+        report.warning(
+            message="Unsupported liquid template",
+            context=f"text {text} liquid_variable {liquid_variable}",
+            exc=e,
+        )
 
     return text
 
@@ -192,15 +202,22 @@
 
     source_config: LookMLSourceConfig
 
-    def __init__(self, source_config: LookMLSourceConfig):
+    def __init__(
+        self,
+        source_config: LookMLSourceConfig,
+        reporter: LookMLSourceReport,
+        manifest_lookml_constant: Optional[List[Dict[str, str]]] = [],
+    ):
         self.source_config = source_config
+        self.reporter = reporter
+        self.manifest_lookml_constant = manifest_lookml_constant
 
     def transform(self, view: dict) -> dict:
         value_to_transform: Optional[str] = None
 
-        # is_attribute_supported check is required because not all transformer works on all attributes in current
-        # case mostly all transformer works on sql_table_name and derived.sql attributes,
-        # however IncompleteSqlTransformer only transform the derived.sql attribute
+        # is_attribute_supported check is required because not all transformers work on all attributes in the current
+        #  case, mostly all transformers work on sql_table_name and derived.sql attributes;
+        # however, IncompleteSqlTransformer only transform the derived.sql attribute
         if SQL_TABLE_NAME in view and self.is_attribute_supported(SQL_TABLE_NAME):
             # Give precedence to already processed transformed view.sql_table_name to apply more transformation
             value_to_transform = view.get(
@@ -252,7 +269,8 @@
     def _apply_transformation(self, value: str, view: dict) -> str:
         return resolve_liquid_variable(
             text=value,
-            liquid_variable=self.source_config.liquid_variable,
+            liquid_variable=self.source_config.liquid_variables,
+            report=self.reporter,
         )
 
 
@@ -287,7 +305,7 @@
 
 class DropDerivedViewPatternTransformer(LookMLViewTransformer):
     """
-    drop ${} from datahub_transformed_sql_table_name and  view["derived_table"]["datahub_transformed_sql_table_name"] values.
+    drop ${} from datahub_transformed_sql_table_name and view["derived_table"]["datahub_transformed_sql_table_name"] values.
 
     Example: transform ${employee_income_source.SQL_TABLE_NAME} to employee_income_source.SQL_TABLE_NAME
     """
@@ -308,8 +326,8 @@
     evaluate_to_true_regx: str
     remove_if_comment_line_regx: str
 
-    def __init__(self, source_config: LookMLSourceConfig):
-        super().__init__(source_config=source_config)
+    def __init__(self, source_config: LookMLSourceConfig, reporter: LookMLSourceReport):
+        super().__init__(source_config=source_config, reporter=reporter)
 
         # This regx will keep whatever after -- if looker_environment --
         self.evaluate_to_true_regx = r"-- if {} --".format(
@@ -335,6 +353,56 @@
         return self._apply_regx(value)
 
 
+class LookmlConstantTransformer(LookMLViewTransformer):
+    """
+    Replace LookML constants @{constant} from the manifest/configuration.
+    """
+
+    CONSTANT_PATTERN = r"@{(\w+)}"  # Matches @{constant}
+
+    def resolve_lookml_parameter(self, text: str) -> str:
+        """
+        Resolves LookML constants (@{ }) from manifest or config.
+        Logs warnings for misplaced or missing variables.
+        """
+
+        def replace_constants(match):
+            key = match.group(1)
+            if self.manifest_lookml_constant:
+                value = next(
+                    (
+                        item["value"]
+                        for item in self.manifest_lookml_constant
+                        if item["name"] == key
+                    ),
+                    None,
+                )
+                if value:
+                    return value
+
+            # Resolve constant from config
+            if key in self.source_config.lookml_constants:
+                return str(self.source_config.lookml_constants.get(key))
+
+            # Check if it's a misplaced lookml constant
+            if key in self.source_config.liquid_variables:
+                self.reporter.report_warning(
+                    title="Misplaced lookml constant",
+                    message="Misplaced lookml constant, Use 'lookml_constants' instead of 'liquid_variables'.",
+                    context=f"Key {key}",
+                )
+                return f"@{{{key}}}"
+
+            logger.warning(f"Constant '@{{{key}}}' not found in configuration.")
+            return "NULL"
+
+        # Resolve @{} (constant)
+        return re.sub(self.CONSTANT_PATTERN, replace_constants, text)
+
+    def _apply_transformation(self, value: str, view: dict) -> str:
+        return self.resolve_lookml_parameter(text=value)
+
+
 class TransformedLookMlView:
     """
     TransformedLookMlView is collecting output of LookMLViewTransformer and creating a new transformed LookML view.
@@ -390,22 +458,29 @@
 def process_lookml_template_language(
     source_config: LookMLSourceConfig,
     view_lkml_file_dict: dict,
+    reporter: LookMLSourceReport,
+    manifest_lookml_constant: Optional[List[Dict[str, str]]],
 ) -> None:
     if "views" not in view_lkml_file_dict:
         return
 
     transformers: List[LookMLViewTransformer] = [
         LookMlIfCommentTransformer(
-            source_config=source_config
+            source_config=source_config, reporter=reporter
         ),  # First evaluate the -- if -- comments. Looker does the same
         LiquidVariableTransformer(
-            source_config=source_config
+            source_config=source_config, reporter=reporter
         ),  # Now resolve liquid variables
+        LookmlConstantTransformer(
+            source_config=source_config,
+            manifest_lookml_constant=manifest_lookml_constant,
+            reporter=reporter,
+        ),  # Resolve @{} constant with its corresponding value
         DropDerivedViewPatternTransformer(
-            source_config=source_config
+            source_config=source_config, reporter=reporter
         ),  # Remove any ${} symbol
         IncompleteSqlTransformer(
-            source_config=source_config
+            source_config=source_config, reporter=reporter
         ),  # complete any incomplete sql
     ]
 
@@ -422,12 +497,16 @@
 def load_and_preprocess_file(
     path: Union[str, pathlib.Path],
     source_config: LookMLSourceConfig,
+    reporter: LookMLSourceReport,
+    manifest_lookml_constant: Optional[List[Dict[str, str]]] = [],
 ) -> dict:
     parsed = load_lkml(path)
 
     process_lookml_template_language(
         view_lkml_file_dict=parsed,
+        reporter=reporter,
         source_config=source_config,
+        manifest_lookml_constant=manifest_lookml_constant,
     )
 
     return parsed

metadata-ingestion/src/datahub/ingestion/source/looker/lookml_config.py

@@ -158,13 +158,27 @@ class LookMLSourceConfig(
         description="When enabled, looker refinement will be processed to adapt an existing view.",
     )
 
-    liquid_variable: Dict[Any, Any] = Field(
+    liquid_variables: Dict[Any, Any] = Field(
         {},
-        description="A dictionary containing Liquid variables and their corresponding values, utilized in SQL-defined "
+        description="A dictionary containing Liquid variables, Liquid logic, and LookML parameters with their corresponding values, utilized in SQL-defined "
         "derived views. The Liquid template will be resolved in view.derived_table.sql and "
         "view.sql_table_name. Defaults to an empty dictionary.",
     )
 
+    _liquid_variable_deprecated = pydantic_renamed_field(
+        old_name="liquid_variable", new_name="liquid_variables", print_warning=True
+    )
+
+    lookml_constants: Dict[str, str] = Field(
+        {},
+        description=(
+            "A dictionary containing LookML constants (`@{constant_name}`) and their values. "
+            "If a constant is defined in the `manifest.lkml` file, its value will be used. "
+            "If not found in the manifest, the value from this config will be used instead. "
+            "Defaults to an empty dictionary."
+        ),
+    )
+
     looker_environment: Literal["prod", "dev"] = Field(
         "prod",
         description="A looker prod or dev environment. "