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_variable:
+    #   _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_constant:
+    #   star_award_winner_year: "public.winner_2025"
+
 # sink configs

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

@@ -1,11 +1,31 @@
 #### 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
-    ```
+
+   ```yml
+   liquid_variable:
+     user_attributes:
+       db: ANALYTICS_PROD
+   ```
+
+2. If a view contains a LookML constant (e.g., `sql_table_name: @{db}.kafka_streaming.events;`), its value will be resolved by first checking the manifest file and, if not found, then checking the config as shown below:.
+
+   ```yml
+   lookml_constant:
+     db: ANALYTICS_PROD
+   ```
+
+   **Example**:
+
+   ```lkml
+   sql_table_name: @{db}.kafka_streaming.events;;
+   ```
+
+   **Resolved Output**:
+
+   ```
+   ANALYTICS_PROD.kafka_streaming.events
+   ```
 
 ### Multi-Project LookML (Advanced)
 

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,
+        looker_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.looker_constant = looker_constant
 
     def _load_viewfile(
         self, project_name: str, path: str, reporter: LookMLSourceReport
@@ -74,6 +76,7 @@ def _load_viewfile(
             parsed = load_and_preprocess_file(
                 path=path,
                 source_config=self.source_config,
+                looker_constant=self.looker_constant,
             )
 
             looker_viewfile = LookerViewFile.from_looker_dict(

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

@@ -192,15 +192,20 @@
 
     source_config: LookMLSourceConfig
 
-    def __init__(self, source_config: LookMLSourceConfig):
+    def __init__(
+        self,
+        source_config: LookMLSourceConfig,
+        looker_constant: Optional[List[Dict[str, str]]] = [],
+    ):
         self.source_config = source_config
+        self.looker_constant = looker_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(
@@ -287,7 +292,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
     """
@@ -335,6 +340,54 @@
         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.looker_constant:
+                value = next(
+                    (
+                        item["value"]
+                        for item in self.looker_constant
+                        if item["name"] == key
+                    ),
+                    None,
+                )
+                if value:
+                    return value
+
+            # Resolve constant
+            if key in self.source_config.lookml_constant:
+                return str(self.source_config.lookml_constant.get(key))
+
+            # Check if it's a misplaced liquid variable
+            if key in self.source_config.liquid_variable:
+                logger.warning(
+                    f"Misplaced liquid variable '@{{{key}}}' detected. Use '${{{key}}}' instead."
+                )
+                return f"@{{{key}}}"
+
+            logger.warning(f"Parameter '@{{{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,6 +443,7 @@
 def process_lookml_template_language(
     source_config: LookMLSourceConfig,
     view_lkml_file_dict: dict,
+    looker_constant: Optional[List[Dict[str, str]]],
 ) -> None:
     if "views" not in view_lkml_file_dict:
         return
@@ -401,6 +455,9 @@
         LiquidVariableTransformer(
             source_config=source_config
         ),  # Now resolve liquid variables
+        LookmlConstantTransformer(
+            source_config=source_config, looker_constant=looker_constant
+        ),  # Resolve @{} constant with its corresponding value
         DropDerivedViewPatternTransformer(
             source_config=source_config
         ),  # Remove any ${} symbol
@@ -422,12 +479,14 @@
 def load_and_preprocess_file(
     path: Union[str, pathlib.Path],
     source_config: LookMLSourceConfig,
+    looker_constant: Optional[List[Dict[str, str]]] = [],
 ) -> dict:
     parsed = load_lkml(path)
 
     process_lookml_template_language(
         view_lkml_file_dict=parsed,
         source_config=source_config,
+        looker_constant=looker_constant,
     )
 
     return parsed

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

@@ -160,11 +160,20 @@ class LookMLSourceConfig(
 
     liquid_variable: 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.",
     )
 
+    lookml_constant: Dict[str, str] = Field(
+        {},
+        description=(
+            "A dictionary containing LookML constant (`@{constant_name}`) and their values. "
+            "Parameters are used to define dynamic references in LookML views, such as `view.sql_table_name`."
+            "Defaults to an empty dictionary."
+        ),
+    )
+
     looker_environment: Literal["prod", "dev"] = Field(
         "prod",
         description="A looker prod or dev environment. "

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

@@ -253,6 +253,7 @@ class LookerManifest:
     # This must be set if the manifest has local_dependency entries.
     # See https://cloud.google.com/looker/docs/reference/param-manifest-project-name
     project_name: Optional[str]
+    constants: Optional[List[Dict[str, str]]]
 
     local_dependencies: List[str]
     remote_dependencies: List[LookerRemoteDependency]
@@ -309,6 +310,8 @@ def __init__(self, config: LookMLSourceConfig, ctx: PipelineContext):
                     "manage_models permission enabled on this API key."
                 ) from err
 
+        self.looker_constant: List[Dict[str, str]] = []
+
     def _load_model(self, path: str) -> LookerModel:
         logger.debug(f"Loading model from file {path}")
 
@@ -506,6 +509,7 @@ def get_manifest_if_present(self, folder: pathlib.Path) -> Optional[LookerManife
 
             manifest = LookerManifest(
                 project_name=manifest_dict.get("project_name"),
+                constants=manifest_dict.get("constants", []),
                 local_dependencies=[
                     x["project"] for x in manifest_dict.get("local_dependencys", [])
                 ],
@@ -574,7 +578,7 @@ def get_workunits_internal(self) -> Iterable[MetadataWorkUnit]:
                 self.base_projects_folder[project] = p_ref
 
             self._recursively_check_manifests(
-                tmp_dir, BASE_PROJECT_NAME, visited_projects
+                tmp_dir, BASE_PROJECT_NAME, visited_projects, self.looker_constant
             )
 
             yield from self.get_internal_workunits()
@@ -587,7 +591,11 @@ def get_workunits_internal(self) -> Iterable[MetadataWorkUnit]:
                 )
 
     def _recursively_check_manifests(
-        self, tmp_dir: str, project_name: str, project_visited: Set[str]
+        self,
+        tmp_dir: str,
+        project_name: str,
+        project_visited: Set[str],
+        looker_constant: List[Dict[str, str]],
     ) -> None:
         if project_name in project_visited:
             return
@@ -604,6 +612,9 @@ def _recursively_check_manifests(
         if not manifest:
             return
 
+        if manifest.constants:
+            looker_constant.extend(manifest.constants)
+
         # Special case handling if the root project has a name in the manifest file.
         if project_name == BASE_PROJECT_NAME and manifest.project_name:
             if (
@@ -663,20 +674,22 @@ def _recursively_check_manifests(
                 project_visited.add(project_name)
             else:
                 self._recursively_check_manifests(
-                    tmp_dir, remote_project.name, project_visited
+                    tmp_dir, remote_project.name, project_visited, looker_constant
                 )
 
         for project in manifest.local_dependencies:
-            self._recursively_check_manifests(tmp_dir, project, project_visited)
+            self._recursively_check_manifests(
+                tmp_dir, project, project_visited, looker_constant
+            )
 
     def get_internal_workunits(self) -> Iterable[MetadataWorkUnit]:  # noqa: C901
         assert self.source_config.base_folder
-
         viewfile_loader = LookerViewFileLoader(
             self.source_config.project_name,
             self.base_projects_folder,
             self.reporter,
             self.source_config,
+            self.looker_constant,
         )
 
         # Some views can be mentioned by multiple 'include' statements and can be included via different connections.

metadata-ingestion/tests/integration/lookml/test_lookml.py

@@ -869,11 +869,15 @@ def test_view_to_view_lineage_and_liquid_template(pytestconfig, tmp_path, mock_t
             "dev_database_prefix": "employee",
             "dev_schema_prefix": "public",
         },
+        "_dev_database_attributes": {
+            "dev_database_prefix": "customer",
+        },
         "dw_eff_dt_date": {
             "_is_selected": True,
         },
         "source_region": "ap-south-1",
     }
+    new_recipe["source"]["config"]["lookml_constant"] = {"winner_table": "dev"}
 
     pipeline = Pipeline.create(new_recipe)
     pipeline.run()

metadata-ingestion/tests/integration/lookml/vv-lineage-and-liquid-templates/activity_logs.view.lkml

@@ -1,13 +1,6 @@
 view: activity_logs {
   sql_table_name:
-        {% if _user_attributes['looker_env'] == 'dev' %}
-          {{ _user_attributes['dev_database_prefix'] }}analytics.{{ _user_attributes['dev_schema_prefix'] }}staging_app.stg_app__activity_logs
-        {% elsif _user_attributes['looker_env'] == 'prod' %}
-          analytics.staging_app.stg_app__activity_logs
-        {% else %}
-          analytics.staging_app.stg_app__activity_logs
-        {% endif %}
-        ;;
+        {{ _dev_database_attributes['dev_database_prefix'] }}_analytics.{{ _user_attributes['dev_schema_prefix'] }}_staging_app.stg_app__activity_logs;;
 
   dimension: generated_message_id {
     group_label: "IDs"

metadata-ingestion/tests/integration/lookml/vv-lineage-and-liquid-templates/data.model.lkml

@@ -10,6 +10,9 @@ include: "environment_activity_logs.view.lkml"
 include: "employee_income_source_as_per_env.view.lkml"
 include: "rent_as_employee_income_source.view.lkml"
 include: "child_view.view.lkml"
+include: "star_award_winner.view.lkml"
+include: "star_award_winner_dev.view.lkml"
+
 
 explore: activity_logs {
 }
@@ -39,4 +42,10 @@ explore: rent_as_employee_income_source {
 }
 
 explore: child_view {
-}
+}
+
+explore: star_award_winner {
+}
+
+explore: star_award_winner_dev {
+}

metadata-ingestion/tests/integration/lookml/vv-lineage-and-liquid-templates/manifest.lkml

@@ -0,0 +1,15 @@
+constant: customer_support_db {
+  value: "star_award_winner_year"
+  export: none
+}
+
+constant: customer_support_schema {
+  value: "public"
+  export: none
+}
+
+constant: customer_support_table {
+  value: "winner"
+  export: none
+}
+

metadata-ingestion/tests/integration/lookml/vv-lineage-and-liquid-templates/star_award_winner.view.lkml

@@ -0,0 +1,12 @@
+view: star_award_winner {
+  sql_table_name: @{customer_support_db}.@{customer_support_schema}.@{customer_support_table};;
+
+
+  dimension: id {
+    label: "id"
+    primary_key: yes
+    type: number
+    sql: ${TABLE}.id ;;
+  }
+
+}