lakehouse_engine.algorithms.gab

  1"""Module to define Gold Asset Builder algorithm behavior."""
  2
  3import copy
  4from datetime import datetime, timedelta
  5from typing import Union
  6
  7import pendulum
  8from jinja2 import Template
  9from pyspark import Row
 10from pyspark.sql import DataFrame
 11from pyspark.sql.functions import lit
 12
 13from lakehouse_engine.algorithms.algorithm import Algorithm
 14from lakehouse_engine.core.definitions import (
 15    GABCadence,
 16    GABCombinedConfiguration,
 17    GABDefaults,
 18    GABKeys,
 19    GABReplaceableKeys,
 20    GABSpec,
 21    GABStartOfWeek,
 22)
 23from lakehouse_engine.core.exec_env import ExecEnv
 24from lakehouse_engine.core.gab_manager import GABCadenceManager, GABViewManager
 25from lakehouse_engine.core.gab_sql_generator import (
 26    GABDeleteGenerator,
 27    GABInsertGenerator,
 28)
 29from lakehouse_engine.utils.gab_utils import GABPartitionUtils, GABUtils
 30from lakehouse_engine.utils.logging_handler import LoggingHandler
 31
 32
 33class GAB(Algorithm):
 34    """Class representing the gold asset builder."""
 35
 36    _LOGGER = LoggingHandler(__name__).get_logger()
 37    _SPARK_DEFAULT_PARALLELISM_CONFIG = (
 38        "spark.sql.sources.parallelPartitionDiscovery.parallelism"
 39    )
 40    _SPARK_DEFAULT_PARALLELISM_VALUE = "10000"
 41
 42    def __init__(self, acon: dict):
 43        """Construct GAB instances.
 44
 45        Args:
 46            acon: algorithm configuration.
 47        """
 48        self.spec: GABSpec = GABSpec.create_from_acon(acon=acon)
 49
 50    def execute(self) -> None:
 51        """Execute the Gold Asset Builder."""
 52        self._LOGGER.info(f"Reading {self.spec.lookup_table} as lkp_query_builder")
 53        lookup_query_builder_df = ExecEnv.SESSION.read.table(self.spec.lookup_table)
 54        ExecEnv.SESSION.read.table(self.spec.calendar_table).createOrReplaceTempView(
 55            "df_cal"
 56        )
 57        self._LOGGER.info(f"Generating calendar from {self.spec.calendar_table}")
 58
 59        query_label = self.spec.query_label_filter
 60        queue = self.spec.queue_filter
 61        cadence = self.spec.cadence_filter
 62
 63        self._LOGGER.info(f"Query Label Filter {query_label}")
 64        self._LOGGER.info(f"Queue Filter {queue}")
 65        self._LOGGER.info(f"Cadence Filter {cadence}")
 66
 67        gab_path = self.spec.gab_base_path
 68        self._LOGGER.info(f"Gab Base Path {gab_path}")
 69
 70        lookup_query_builder_df = lookup_query_builder_df.filter(
 71            (
 72                (lookup_query_builder_df.query_label.isin(query_label))
 73                & (lookup_query_builder_df.queue.isin(queue))
 74                & (lookup_query_builder_df.is_active != lit("N"))
 75            )
 76        )
 77
 78        lookup_query_builder_df.cache()
 79
 80        for use_case in lookup_query_builder_df.collect():
 81            self._process_use_case(
 82                use_case=use_case,
 83                lookup_query_builder=lookup_query_builder_df,
 84                selected_cadences=cadence,
 85                gab_path=gab_path,
 86            )
 87
 88        lookup_query_builder_df.unpersist()
 89
 90    def _process_use_case(
 91        self,
 92        use_case: Row,
 93        lookup_query_builder: DataFrame,
 94        selected_cadences: list[str],
 95        gab_path: str,
 96    ) -> None:
 97        """Process each gab use case.
 98
 99        Args:
100            use_case: gab use case to process.
101            lookup_query_builder: gab configuration data.
102            selected_cadences: selected cadences to process.
103            gab_path: gab base path used to get the use case stages sql files.
104        """
105        self._LOGGER.info(f"Executing use case: {use_case['query_label']}")
106
107        reconciliation = GABUtils.get_json_column_as_dict(
108            lookup_query_builder=lookup_query_builder,
109            query_id=use_case["query_id"],
110            query_column="recon_window",
111        )
112        self._LOGGER.info(f"reconcilation window - {reconciliation}")
113        configured_cadences = list(reconciliation.keys())
114
115        stages = GABUtils.get_json_column_as_dict(
116            lookup_query_builder=lookup_query_builder,
117            query_id=use_case["query_id"],
118            query_column="intermediate_stages",
119        )
120        self._LOGGER.info(f"intermediate stages - {stages}")
121
122        self._LOGGER.info(f"selected_cadences: {selected_cadences}")
123        self._LOGGER.info(f"configured_cadences: {configured_cadences}")
124        cadences = self._get_filtered_cadences(selected_cadences, configured_cadences)
125        self._LOGGER.info(f"filtered cadences - {cadences}")
126
127        latest_run_date, latest_config_date = self._get_latest_usecase_data(
128            use_case["query_id"]
129        )
130        self._LOGGER.info(f"latest_config_date: {latest_config_date}")
131        self._LOGGER.info(f"latest_run_date: - {latest_run_date}")
132        self._set_use_case_stage_template_file(stages, gab_path, use_case)
133        processed_cadences = []
134
135        for cadence in cadences:
136            is_cadence_processed = self._process_use_case_query_cadence(
137                cadence,
138                reconciliation,
139                use_case,
140                stages,
141                lookup_query_builder,
142            )
143            if is_cadence_processed:
144                processed_cadences.append(is_cadence_processed)
145
146        if processed_cadences:
147            self._generate_ddl(
148                latest_config_date=latest_config_date,
149                latest_run_date=latest_run_date,
150                query_id=use_case["query_id"],
151                lookup_query_builder=lookup_query_builder,
152            )
153        else:
154            self._LOGGER.info(
155                f"Skipping use case {use_case['query_label']}. No cadence processed "
156                "for the use case."
157            )
158
159    @classmethod
160    def _set_use_case_stage_template_file(
161        cls, stages: dict, gab_path: str, use_case: Row
162    ) -> None:
163        """Set templated file for each stage.
164
165        Args:
166            stages: use case stages with their configuration.
167            gab_path: gab base path used to get the use case stages SQL files.
168            use_case: gab use case to process.
169        """
170        cls._LOGGER.info("Reading templated file for each stage...")
171
172        for i in range(1, len(stages) + 1):
173            stage = stages[str(i)]
174            stage_file_path = stage["file_path"]
175            full_path = gab_path + stage_file_path
176            cls._LOGGER.info(f"Stage file path is: {full_path}")
177            file_read = open(full_path, "r").read()
178            templated_file = file_read.replace(
179                "replace_offset_value", str(use_case["timezone_offset"])
180            )
181            stage["templated_file"] = templated_file
182            stage["full_file_path"] = full_path
183
184    def _process_use_case_query_cadence(
185        self,
186        cadence: str,
187        reconciliation: dict,
188        use_case: Row,
189        stages: dict,
190        lookup_query_builder: DataFrame,
191    ) -> bool:
192        """Identify use case reconciliation window and cadence.
193
194        Args:
195            cadence:  cadence to process.
196            reconciliation: configured use case reconciliation window.
197            use_case: gab use case to process.
198            stages: use case stages with their configuration.
199            lookup_query_builder: gab configuration data.
200        """
201        selected_reconciliation_window = {}
202        selected_cadence = reconciliation.get(cadence)
203        self._LOGGER.info(f"Processing cadence: {cadence}")
204        self._LOGGER.info(f"Reconciliation Window - {selected_cadence}")
205
206        if selected_cadence:
207            selected_reconciliation_window = selected_cadence.get("recon_window")
208
209        self._LOGGER.info(f"{cadence}: {self.spec.start_date} - {self.spec.end_date}")
210
211        start_of_week = use_case["start_of_the_week"]
212
213        self._set_week_configuration_by_uc_start_of_week(start_of_week)
214
215        cadence_configuration_at_end_date = (
216            GABUtils.get_cadence_configuration_at_end_date(self.spec.end_date)
217        )
218
219        reconciliation_cadences = GABUtils().get_reconciliation_cadences(
220            cadence=cadence,
221            selected_reconciliation_window=selected_reconciliation_window,
222            cadence_configuration_at_end_date=cadence_configuration_at_end_date,
223            rerun_flag=self.spec.rerun_flag,
224        )
225
226        start_date_str = GABUtils.format_datetime_to_default(self.spec.start_date)
227        end_date_str = GABUtils.format_datetime_to_default(self.spec.end_date)
228
229        for reconciliation_cadence, snapshot_flag in reconciliation_cadences.items():
230            self._process_reconciliation_cadence(
231                reconciliation_cadence=reconciliation_cadence,
232                snapshot_flag=snapshot_flag,
233                cadence=cadence,
234                start_date_str=start_date_str,
235                end_date_str=end_date_str,
236                use_case=use_case,
237                lookup_query_builder=lookup_query_builder,
238                stages=stages,
239            )
240
241        return (cadence in reconciliation.keys()) or (
242            reconciliation_cadences is not None
243        )
244
245    def _process_reconciliation_cadence(
246        self,
247        reconciliation_cadence: str,
248        snapshot_flag: str,
249        cadence: str,
250        start_date_str: str,
251        end_date_str: str,
252        use_case: Row,
253        lookup_query_builder: DataFrame,
254        stages: dict,
255    ) -> None:
256        """Process use case reconciliation window.
257
258        Reconcile the pre-aggregated data to cover the late events.
259
260        Args:
261            reconciliation_cadence: reconciliation to process.
262            snapshot_flag: flag indicating if for this cadence the snapshot is enabled.
263            cadence: cadence to process.
264            start_date_str: start date of the period to process.
265            end_date_str: end date of the period to process.
266            use_case: gab use case to process.
267            lookup_query_builder: gab configuration data.
268            stages: use case stages with their configuration.
269
270        Example:
271            Cadence: week;
272            Reconciliation: monthly;
273            This means every weekend previous week aggregations will be calculated and
274                on month end we will reconcile the numbers calculated for last 4 weeks
275                to readjust the number for late events.
276        """
277        (
278            window_start_date,
279            window_end_date,
280            filter_start_date,
281            filter_end_date,
282        ) = GABCadenceManager().extended_window_calculator(
283            cadence,
284            reconciliation_cadence,
285            self.spec.current_date,
286            start_date_str,
287            end_date_str,
288            use_case["query_type"],
289            self.spec.rerun_flag,
290            snapshot_flag,
291        )
292
293        if use_case["timezone_offset"]:
294            filter_start_date = filter_start_date + timedelta(
295                hours=use_case["timezone_offset"]
296            )
297            filter_end_date = filter_end_date + timedelta(
298                hours=use_case["timezone_offset"]
299            )
300
301        filter_start_date_str = GABUtils.format_datetime_to_default(filter_start_date)
302        filter_end_date_str = GABUtils.format_datetime_to_default(filter_end_date)
303
304        partition_end = GABUtils.format_datetime_to_default(
305            (window_end_date - timedelta(days=1))
306        )
307
308        window_start_date_str = GABUtils.format_datetime_to_default(window_start_date)
309        window_end_date_str = GABUtils.format_datetime_to_default(window_end_date)
310
311        partition_filter = GABPartitionUtils.get_partition_condition(
312            filter_start_date_str, partition_end
313        )
314
315        self._LOGGER.info(
316            "extended window for start and end dates are: "
317            f"{filter_start_date_str} - {filter_end_date_str}"
318        )
319
320        unpersist_list = []
321
322        for i in range(1, len(stages) + 1):
323            stage = stages[str(i)]
324            templated_file = stage["templated_file"]
325            stage_file_path = stage["full_file_path"]
326
327            templated = self._process_use_case_query_step(
328                stage=stages[str(i)],
329                templated_file=templated_file,
330                use_case=use_case,
331                reconciliation_cadence=reconciliation_cadence,
332                cadence=cadence,
333                snapshot_flag=snapshot_flag,
334                window_start_date=window_start_date_str,
335                partition_end=partition_end,
336                filter_start_date=filter_start_date_str,
337                filter_end_date=filter_end_date_str,
338                partition_filter=partition_filter,
339            )
340
341            temp_stage_view_name = self._create_stage_view(
342                templated,
343                stages[str(i)],
344                window_start_date_str,
345                window_end_date_str,
346                use_case["query_id"],
347                use_case["query_label"],
348                cadence,
349                stage_file_path,
350            )
351            unpersist_list.append(temp_stage_view_name)
352
353        insert_success = self._generate_view_statement(
354            query_id=use_case["query_id"],
355            cadence=cadence,
356            temp_stage_view_name=temp_stage_view_name,
357            lookup_query_builder=lookup_query_builder,
358            window_start_date=window_start_date_str,
359            window_end_date=window_end_date_str,
360            query_label=use_case["query_label"],
361        )
362        self._LOGGER.info(f"Inserted data to generate the view: {insert_success}")
363
364        self._unpersist_cached_views(unpersist_list)
365
366    def _process_use_case_query_step(
367        self,
368        stage: dict,
369        templated_file: str,
370        use_case: Row,
371        reconciliation_cadence: str,
372        cadence: str,
373        snapshot_flag: str,
374        window_start_date: str,
375        partition_end: str,
376        filter_start_date: str,
377        filter_end_date: str,
378        partition_filter: str,
379    ) -> str:
380        """Process each use case step.
381
382        Process any intermediate view defined in the gab configuration table as step for
383            the use case.
384
385        Args:
386            stage: stage to process.
387            templated_file: sql file to process at this stage.
388            use_case: gab use case to process.
389            reconciliation_cadence: configured use case reconciliation window.
390            cadence: cadence to process.
391            snapshot_flag: flag indicating if for this cadence the snapshot is enabled.
392            window_start_date: start date for the configured stage.
393            partition_end: end date for the configured stage.
394            filter_start_date: filter start date to replace in the stage query.
395            filter_end_date: filter end date to replace in the stage query.
396            partition_filter: partition condition.
397        """
398        filter_col = stage["project_date_column"]
399        if stage["filter_date_column"]:
400            filter_col = stage["filter_date_column"]
401
402        # dummy value to avoid empty error if empty on the configuration
403        project_col = stage.get("project_date_column", "X")
404
405        gab_base_configuration_copy = copy.deepcopy(
406            GABCombinedConfiguration.COMBINED_CONFIGURATION.value
407        )
408
409        for item in gab_base_configuration_copy.values():
410            self._update_rendered_item_cadence(
411                reconciliation_cadence, cadence, project_col, item  # type: ignore
412            )
413
414        (
415            rendered_date,
416            rendered_to_date,
417            join_condition,
418        ) = self._get_cadence_configuration(
419            gab_base_configuration_copy,
420            cadence,
421            reconciliation_cadence,
422            snapshot_flag,
423            use_case["start_of_the_week"],
424            project_col,
425            window_start_date,
426            partition_end,
427        )
428
429        rendered_file = self._render_template_query(
430            templated=templated_file,
431            cadence=cadence,
432            start_of_the_week=use_case["start_of_the_week"],
433            query_id=use_case["query_id"],
434            rendered_date=rendered_date,
435            filter_start_date=filter_start_date,
436            filter_end_date=filter_end_date,
437            filter_col=filter_col,
438            timezone_offset=use_case["timezone_offset"],
439            join_condition=join_condition,
440            partition_filter=partition_filter,
441            rendered_to_date=rendered_to_date,
442        )
443
444        return rendered_file
445
446    @classmethod
447    def _get_filtered_cadences(
448        cls, selected_cadences: list[str], configured_cadences: list[str]
449    ) -> list[str]:
450        """Get filtered cadences.
451
452        Get the intersection of user selected cadences and use case configured cadences.
453
454        Args:
455            selected_cadences: user selected cadences.
456            configured_cadences: use case configured cadences.
457        """
458        return (
459            configured_cadences
460            if "All" in selected_cadences
461            else GABCadence.order_cadences(
462                list(set(selected_cadences).intersection(configured_cadences))
463            )
464        )
465
466    def _get_latest_usecase_data(self, query_id: str) -> tuple[datetime, datetime]:
467        """Get latest use case data.
468
469        Args:
470            query_id: use case query id.
471        """
472        return (
473            self._get_latest_run_date(query_id),
474            self._get_latest_use_case_date(query_id),
475        )
476
477    def _get_latest_run_date(self, query_id: str) -> datetime:
478        """Get latest use case run date.
479
480        Args:
481            query_id: use case query id.
482        """
483        last_success_run_sql = """
484            SELECT run_start_time
485            FROM {database}.gab_log_events
486            WHERE query_id = {query_id}
487            AND stage_name = 'Final Insert'
488            AND status = 'Success'
489            ORDER BY 1 DESC
490            LIMIT 1
491            """.format(  # nosec: B608
492            database=self.spec.target_database, query_id=query_id
493        )
494        try:
495            latest_run_date: datetime = ExecEnv.SESSION.sql(
496                last_success_run_sql
497            ).collect()[0][0]
498        except Exception:
499            latest_run_date = datetime.strptime(
500                "2020-01-01", GABDefaults.DATE_FORMAT.value
501            )
502
503        return latest_run_date
504
505    def _get_latest_use_case_date(self, query_id: str) -> datetime:
506        """Get latest use case configured date.
507
508        Args:
509            query_id: use case query id.
510        """
511        query_config_sql = """
512            SELECT lh_created_on
513            FROM {lkp_query_builder}
514            WHERE query_id = {query_id}
515        """.format(  # nosec: B608
516            lkp_query_builder=self.spec.lookup_table,
517            query_id=query_id,
518        )
519
520        latest_config_date: datetime = ExecEnv.SESSION.sql(query_config_sql).collect()[
521            0
522        ][0]
523
524        return latest_config_date
525
526    @classmethod
527    def _set_week_configuration_by_uc_start_of_week(cls, start_of_week: str) -> None:
528        """Set week configuration by use case start of week.
529
530        Args:
531            start_of_week: use case start of week (MONDAY or SUNDAY).
532        """
533        if start_of_week.upper() == "MONDAY":
534            pendulum.week_starts_at(pendulum.MONDAY)
535            pendulum.week_ends_at(pendulum.SUNDAY)
536        elif start_of_week.upper() == "SUNDAY":
537            pendulum.week_starts_at(pendulum.SUNDAY)
538            pendulum.week_ends_at(pendulum.SATURDAY)
539        else:
540            raise NotImplementedError(
541                f"The requested {start_of_week} is not implemented."
542                "Supported `start_of_week` values: [MONDAY, SUNDAY]"
543            )
544
545    @classmethod
546    def _update_rendered_item_cadence(
547        cls, reconciliation_cadence: str, cadence: str, project_col: str, item: dict
548    ) -> None:
549        """Override item properties based in the rendered item cadence.
550
551        Args:
552            reconciliation_cadence: configured use case reconciliation window.
553            cadence: cadence to process.
554            project_col: use case projection date column name.
555            item: predefined use case combination.
556        """
557        rendered_item = cls._get_rendered_item_cadence(
558            reconciliation_cadence, cadence, project_col, item
559        )
560        item["join_select"] = rendered_item["join_select"]
561        item["project_start"] = rendered_item["project_start"]
562        item["project_end"] = rendered_item["project_end"]
563
564    @classmethod
565    def _get_rendered_item_cadence(
566        cls, reconciliation_cadence: str, cadence: str, project_col: str, item: dict
567    ) -> dict:
568        """Update pre-configured gab parameters with use case data.
569
570        Args:
571            reconciliation_cadence: configured use case reconciliation window.
572            cadence: cadence to process.
573            project_col: use case projection date column name.
574            item: predefined use case combination.
575        """
576        return {
577            GABKeys.JOIN_SELECT: (
578                item[GABKeys.JOIN_SELECT]
579                .replace(GABReplaceableKeys.CONFIG_WEEK_START, "Monday")
580                .replace(
581                    GABReplaceableKeys.RECONCILIATION_CADENCE,
582                    reconciliation_cadence,
583                )
584                .replace(GABReplaceableKeys.CADENCE, cadence)
585            ),
586            GABKeys.PROJECT_START: (
587                item[GABKeys.PROJECT_START]
588                .replace(GABReplaceableKeys.CADENCE, cadence)
589                .replace(GABReplaceableKeys.DATE_COLUMN, project_col)
590            ),
591            GABKeys.PROJECT_END: (
592                item[GABKeys.PROJECT_END]
593                .replace(GABReplaceableKeys.CADENCE, cadence)
594                .replace(GABReplaceableKeys.DATE_COLUMN, project_col)
595            ),
596        }
597
598    @classmethod
599    def _get_cadence_configuration(
600        cls,
601        use_case_configuration: dict,
602        cadence: str,
603        reconciliation_cadence: str,
604        snapshot_flag: str,
605        start_of_week: str,
606        project_col: str,
607        window_start_date: str,
608        partition_end: str,
609    ) -> tuple[str, str, str]:
610        """Get use case configuration fields to replace pre-configured parameters.
611
612        Args:
613            use_case_configuration: use case configuration.
614            cadence: cadence to process.
615            reconciliation_cadence: cadence to be reconciliated.
616            snapshot_flag: flag indicating if for this cadence the snapshot is enabled.
617            start_of_week: use case start of week (MONDAY or SUNDAY).
618            project_col: use case projection date column name.
619            window_start_date: start date for the configured stage.
620            partition_end: end date for the configured stage.
621
622        Returns:
623            rendered_from_date: projection start date.
624            rendered_to_date: projection end date.
625            join_condition: string containing the join condition to replace in the
626                templated query by jinja substitution.
627        """
628        cadence_dict = next(
629            (
630                dict(configuration)
631                for configuration in use_case_configuration.values()
632                if (
633                    (cadence in configuration["cadence"])
634                    and (reconciliation_cadence in configuration["recon"])
635                    and (snapshot_flag in configuration["snap_flag"])
636                    and (
637                        GABStartOfWeek.get_start_of_week()[start_of_week.upper()]
638                        in configuration["week_start"]
639                    )
640                )
641            ),
642            None,
643        )
644        rendered_from_date = None
645        rendered_to_date = None
646        join_condition = None
647
648        if cadence_dict:
649            rendered_from_date = (
650                cadence_dict[GABKeys.PROJECT_START]
651                .replace(GABReplaceableKeys.CADENCE, cadence)
652                .replace(GABReplaceableKeys.DATE_COLUMN, project_col)
653            )
654            rendered_to_date = (
655                cadence_dict[GABKeys.PROJECT_END]
656                .replace(GABReplaceableKeys.CADENCE, cadence)
657                .replace(GABReplaceableKeys.DATE_COLUMN, project_col)
658            )
659
660            if cadence_dict[GABKeys.JOIN_SELECT]:
661                join_condition = """
662                 inner join (
663                     {join_select} from df_cal
664                     where calendar_date
665                     between '{bucket_start}' and '{bucket_end}'
666                 )
667                 df_cal on date({date_column})
668                     between df_cal.cadence_start_date and df_cal.cadence_end_date
669                 """.format(
670                    join_select=cadence_dict[GABKeys.JOIN_SELECT],
671                    bucket_start=window_start_date,
672                    bucket_end=partition_end,
673                    date_column=project_col,
674                )
675
676        return rendered_from_date, rendered_to_date, join_condition
677
678    def _render_template_query(
679        self,
680        templated: str,
681        cadence: str,
682        start_of_the_week: str,
683        query_id: str,
684        rendered_date: str,
685        filter_start_date: str,
686        filter_end_date: str,
687        filter_col: str,
688        timezone_offset: str,
689        join_condition: str,
690        partition_filter: str,
691        rendered_to_date: str,
692    ) -> str:
693        """Replace jinja templated parameters in the SQL with the actual data.
694
695        Args:
696            templated: templated sql file to process at this stage.
697            cadence: cadence to process.
698            start_of_the_week: use case start of week (MONDAY or SUNDAY).
699            query_id: gab configuration table use case identifier.
700            rendered_date: projection start date.
701            filter_start_date: filter start date to replace in the stage query.
702            filter_end_date: filter end date to replace in the stage query.
703            filter_col: use case projection date column name.
704            timezone_offset: timezone offset configured in the use case.
705            join_condition: string containing the join condition.
706            partition_filter: partition condition.
707            rendered_to_date: projection end date.
708        """
709        return Template(templated).render(
710            cadence="'{cadence}' as cadence".format(cadence=cadence),
711            cadence_run=cadence,
712            week_start=start_of_the_week,
713            query_id="'{query_id}' as query_id".format(query_id=query_id),
714            project_date_column=rendered_date,
715            target_table=self.spec.target_table,
716            database=self.spec.source_database,
717            start_date=filter_start_date,
718            end_date=filter_end_date,
719            filter_date_column=filter_col,
720            offset_value=timezone_offset,
721            joins=join_condition if join_condition else "",
722            partition_filter=partition_filter,
723            to_date=rendered_to_date,
724        )
725
726    def _create_stage_view(
727        self,
728        rendered_template: str,
729        stage: dict,
730        window_start_date: str,
731        window_end_date: str,
732        query_id: str,
733        query_label: str,
734        cadence: str,
735        stage_file_path: str,
736    ) -> str:
737        """Create each use case stage view.
738
739        Each stage has a specific order and refer to a specific SQL to be executed.
740
741        Args:
742            rendered_template: rendered stage SQL file.
743            stage: stage to process.
744            window_start_date: start date for the configured stage.
745            window_end_date: end date for the configured stage.
746            query_id: gab configuration table use case identifier.
747            query_label: gab configuration table use case name.
748            cadence: cadence to process.
749            stage_file_path: full stage file path (gab path + stage path).
750        """
751        run_start_time = datetime.now()
752        creation_status: str
753        error_message: Union[Exception, str]
754
755        try:
756            tmp = ExecEnv.SESSION.sql(rendered_template)
757            num_partitions = ExecEnv.SESSION.conf.get(
758                self._SPARK_DEFAULT_PARALLELISM_CONFIG,
759                self._SPARK_DEFAULT_PARALLELISM_VALUE,
760            )
761
762            if stage["repartition"]:
763                if stage["repartition"].get("numPartitions"):
764                    num_partitions = stage["repartition"]["numPartitions"]
765
766                if stage["repartition"].get("keys"):
767                    tmp = tmp.repartition(
768                        int(num_partitions), *stage["repartition"]["keys"]
769                    )
770                    self._LOGGER.info("Repartitioned on given Key(s)")
771                else:
772                    tmp = tmp.repartition(int(num_partitions))
773                    self._LOGGER.info("Repartitioned on given partition count")
774
775            temp_step_view_name: str = stage["table_alias"]
776            tmp.createOrReplaceTempView(temp_step_view_name)
777
778            if stage["storage_level"]:
779                ExecEnv.SESSION.sql(
780                    "CACHE TABLE {tbl} "
781                    "OPTIONS ('storageLevel' '{type}')".format(
782                        tbl=temp_step_view_name,
783                        type=stage["storage_level"],
784                    )
785                )
786                ExecEnv.SESSION.sql(
787                    "SELECT COUNT(*) FROM {tbl}".format(  # nosec: B608
788                        tbl=temp_step_view_name
789                    )
790                )
791                self._LOGGER.info(f"Cached stage view - {temp_step_view_name} ")
792
793            creation_status = "Success"
794            error_message = "NA"
795        except Exception as err:
796            creation_status = "Failed"
797            error_message = err
798            raise err
799        finally:
800            run_end_time = datetime.now()
801            GABUtils().logger(
802                run_start_time,
803                run_end_time,
804                window_start_date,
805                window_end_date,
806                query_id,
807                query_label,
808                cadence,
809                stage_file_path,
810                rendered_template,
811                creation_status,
812                error_message,
813                self.spec.target_database,
814            )
815
816        return temp_step_view_name
817
818    def _generate_view_statement(
819        self,
820        query_id: str,
821        cadence: str,
822        temp_stage_view_name: str,
823        lookup_query_builder: DataFrame,
824        window_start_date: str,
825        window_end_date: str,
826        query_label: str,
827    ) -> bool:
828        """Feed use case data to the insights table (default: unified use case table).
829
830        Args:
831            query_id: gab configuration table use case identifier.
832            cadence: cadence to process.
833            temp_stage_view_name: name of the temp view generated by the stage.
834            lookup_query_builder: gab configuration data.
835            window_start_date: start date for the configured stage.
836            window_end_date: end date for the configured stage.
837            query_label: gab configuration table use case name.
838        """
839        run_start_time = datetime.now()
840        creation_status: str
841        error_message: Union[Exception, str]
842
843        GABDeleteGenerator(
844            query_id=query_id,
845            cadence=cadence,
846            temp_stage_view_name=temp_stage_view_name,
847            lookup_query_builder=lookup_query_builder,
848            target_database=self.spec.target_database,
849            target_table=self.spec.target_table,
850        ).generate_sql()
851
852        gen_ins = GABInsertGenerator(
853            query_id=query_id,
854            cadence=cadence,
855            final_stage_table=temp_stage_view_name,
856            lookup_query_builder=lookup_query_builder,
857            target_database=self.spec.target_database,
858            target_table=self.spec.target_table,
859        ).generate_sql()
860        try:
861            ExecEnv.SESSION.sql(gen_ins)
862
863            creation_status = "Success"
864            error_message = "NA"
865            inserted = True
866        except Exception as err:
867            creation_status = "Failed"
868            error_message = err
869            raise
870        finally:
871            run_end_time = datetime.now()
872            GABUtils().logger(
873                run_start_time,
874                run_end_time,
875                window_start_date,
876                window_end_date,
877                query_id,
878                query_label,
879                cadence,
880                "Final Insert",
881                gen_ins,
882                creation_status,
883                error_message,
884                self.spec.target_database,
885            )
886
887        return inserted
888
889    @classmethod
890    def _unpersist_cached_views(cls, unpersist_list: list[str]) -> None:
891        """Unpersist cached views.
892
893        Args:
894            unpersist_list: list containing the view names to unpersist.
895        """
896        [
897            ExecEnv.SESSION.sql("UNCACHE TABLE {tbl}".format(tbl=i))
898            for i in unpersist_list
899        ]
900
901    def _generate_ddl(
902        self,
903        latest_config_date: datetime,
904        latest_run_date: datetime,
905        query_id: str,
906        lookup_query_builder: DataFrame,
907    ) -> None:
908        """Generate the actual gold asset.
909
910        It will create and return the view containing all specified dimensions, metrics
911            and computed metric for each cadence/reconciliation window.
912
913        Args:
914            latest_config_date: latest use case configuration date.
915            latest_run_date: latest use case run date.
916            query_id: gab configuration table use case identifier.
917            lookup_query_builder: gab configuration data.
918        """
919        if str(latest_config_date) > str(latest_run_date):
920            GABViewManager(
921                query_id=query_id,
922                lookup_query_builder=lookup_query_builder,
923                target_database=self.spec.target_database,
924                target_table=self.spec.target_table,
925            ).generate_use_case_views()
926        else:
927            self._LOGGER.info(
928                "View is not being re-created as there are no changes in the "
929                "configuration after the latest run"
930            )