Merge branch 'main' into meetup-connector

Author: akashverma0786

bootstrap/sql/migrations/native/1.13.0/mysql/schemaChanges.sql

@@ -216,6 +216,42 @@ WHERE pipelineType = 'profiler'
     OR JSON_CONTAINS_PATH(json, 'one', '$.sourceConfig.config.profileSampleType')
     OR JSON_CONTAINS_PATH(json, 'one', '$.sourceConfig.config.samplingMethodType'));
 
+-- ingestion_pipeline_entity (testSuite pipelines): build profileSampleConfig (skip if already migrated)
+UPDATE ingestion_pipeline_entity
+SET json = JSON_SET(
+    json,
+    '$.sourceConfig.config.profileSampleConfig',
+    JSON_OBJECT(
+        'sampleConfigType', 'STATIC',
+        'config', JSON_OBJECT(
+            'profileSample', JSON_EXTRACT(json, '$.sourceConfig.config.profileSample'),
+            'profileSampleType', COALESCE(
+                JSON_EXTRACT(json, '$.sourceConfig.config.profileSampleType'),
+                CAST('"PERCENTAGE"' AS JSON)
+            ),
+            'samplingMethodType', JSON_EXTRACT(json, '$.sourceConfig.config.samplingMethodType')
+        )
+    )
+)
+WHERE pipelineType = 'testSuite'
+  AND JSON_EXTRACT(json, '$.sourceConfig.config.profileSample') IS NOT NULL
+  AND JSON_TYPE(JSON_EXTRACT(json, '$.sourceConfig.config.profileSample')) != 'NULL'
+  AND NOT JSON_CONTAINS_PATH(json, 'one', '$.sourceConfig.config.profileSampleConfig');
+
+-- ingestion_pipeline_entity (testSuite pipelines): remove old flat fields
+UPDATE ingestion_pipeline_entity
+SET json = JSON_REMOVE(
+    JSON_REMOVE(
+        JSON_REMOVE(json, '$.sourceConfig.config.samplingMethodType'),
+        '$.sourceConfig.config.profileSampleType'
+    ),
+    '$.sourceConfig.config.profileSample'
+)
+WHERE pipelineType = 'testSuite'
+  AND (JSON_CONTAINS_PATH(json, 'one', '$.sourceConfig.config.profileSample')
+    OR JSON_CONTAINS_PATH(json, 'one', '$.sourceConfig.config.profileSampleType')
+    OR JSON_CONTAINS_PATH(json, 'one', '$.sourceConfig.config.samplingMethodType'));
+
 -- RDF distributed indexing state tables
 CREATE TABLE IF NOT EXISTS rdf_index_job (
     id VARCHAR(36) NOT NULL,

bootstrap/sql/migrations/native/1.13.0/postgres/schemaChanges.sql

@@ -235,6 +235,37 @@ WHERE json #>> '{pipelineType}' = 'profiler'
     OR json::jsonb #>> '{sourceConfig,config,profileSampleType}' IS NOT NULL
     OR json::jsonb #>> '{sourceConfig,config,samplingMethodType}' IS NOT NULL);
 
+-- ingestion_pipeline_entity (testSuite pipelines): build profileSampleConfig (skip if already migrated)
+UPDATE ingestion_pipeline_entity
+SET json = jsonb_set(
+    json::jsonb,
+    '{sourceConfig,config,profileSampleConfig}',
+    jsonb_build_object(
+        'sampleConfigType', 'STATIC',
+        'config', jsonb_build_object(
+            'profileSample', json::jsonb #> '{sourceConfig,config,profileSample}',
+            'profileSampleType', COALESCE(
+                json::jsonb #> '{sourceConfig,config,profileSampleType}',
+                '"PERCENTAGE"'::jsonb
+            ),
+            'samplingMethodType', json::jsonb #> '{sourceConfig,config,samplingMethodType}'
+        )
+    )
+)::json
+WHERE json #>> '{pipelineType}' = 'testSuite'
+  AND json::jsonb #>> '{sourceConfig,config,profileSample}' IS NOT NULL
+  AND json::jsonb #> '{sourceConfig,config,profileSampleConfig}' IS NULL;
+
+-- ingestion_pipeline_entity (testSuite pipelines): remove old flat fields
+UPDATE ingestion_pipeline_entity
+SET json = (json::jsonb #- '{sourceConfig,config,profileSample}'
+                        #- '{sourceConfig,config,profileSampleType}'
+                        #- '{sourceConfig,config,samplingMethodType}')::json
+WHERE json #>> '{pipelineType}' = 'testSuite'
+  AND (json::jsonb #>> '{sourceConfig,config,profileSample}' IS NOT NULL
+    OR json::jsonb #>> '{sourceConfig,config,profileSampleType}' IS NOT NULL
+    OR json::jsonb #>> '{sourceConfig,config,samplingMethodType}' IS NOT NULL);
+
 -- RDF distributed indexing state tables
 CREATE TABLE IF NOT EXISTS rdf_index_job (
     id VARCHAR(36) NOT NULL,

ingestion/setup.py

@@ -423,6 +423,8 @@
     # TODO: Remove once no unit test requires testcontainers
     "testcontainers",
     VERSIONS["factory-boy"],
+    *plugins["exasol"],
+    *plugins["teradata"],
 }
 
 test = {
@@ -491,6 +493,8 @@
     VERSIONS["kafka-connect"],
     VERSIONS["factory-boy"],
     "locust~=2.32.0",
+    *plugins["exasol"],
+    *plugins["teradata"],
 }
 
 docs = {

ingestion/src/metadata/data_quality/runner/base_test_suite_source.py

@@ -34,10 +34,7 @@
 from metadata.generated.schema.type.entityReference import EntityReference
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.sampler.models import (
-    ProfileSampleConfig,
-    ProfileSampleConfigType,
     SampleConfig,
-    StaticSamplingConfig,
 )
 from metadata.sampler.sampler_interface import SamplerInterface  # noqa: TC001
 from metadata.utils.bigquery_utils import copy_service_config
@@ -126,15 +123,8 @@ def create_data_quality_interface(self) -> TestSuiteInterface:
             schema_entity=schema_entity,
             database_entity=database_entity,
             default_sample_config=SampleConfig(
-                profileSampleConfig=ProfileSampleConfig(
-                    sampleConfigType=ProfileSampleConfigType.STATIC,
-                    config=StaticSamplingConfig(
-                        profileSample=self.source_config.profileSample,
-                        profileSampleType=self.source_config.profileSampleType,
-                        samplingMethodType=self.source_config.samplingMethodType,
-                    ),
-                )
-                if self.source_config.profileSample
+                profileSampleConfig=self.source_config.profileSampleConfig
+                if self.source_config.profileSampleConfig
                 else None,
             ),
         )

ingestion/src/metadata/data_quality/validations/column/sqlalchemy/columnValuesToBeUnique.py

@@ -75,7 +75,7 @@ def _run_results(self, metric: Metrics, column: Column) -> Optional[int]:  # noq
         """
         count = Metrics.valuesCount.value(column).fn()
         grouped_cte = (
-            select(count.label(column.name)).select_from(self.runner.dataset).group_by(column).cte("grouped_cte")
+            select(count.label(column.name)).select_from(self.runner.dataset).group_by(column).cte("grouped_cte")  # type: ignore
         )
         unique_count = Metrics.uniqueCount.value(column).query(
             sample=self.runner.dataset,

ingestion/src/metadata/data_quality/validations/table/sqlalchemy/tableDiff.py

@@ -56,6 +56,7 @@
 from metadata.profiler.orm.functions.md5 import MD5
 from metadata.profiler.orm.functions.substr import Substr
 from metadata.profiler.orm.registry import Dialects, PythonDialects
+from metadata.sampler.config import resolve_static_sampling_config
 from metadata.utils.collections import CaseInsensitiveList
 from metadata.utils.credentials import normalize_pem_string
 from metadata.utils.logger import test_suite_logger
@@ -443,9 +444,12 @@ def sample_where_clause(self) -> Tuple[Optional[str], Optional[str]]:  # noqa: U
             return None, None
         profile_sample_config = config.profileSampleConfig if config else None
         sample_config = profile_sample_config.root if profile_sample_config else None
-        static = sample_config.config if sample_config else None
-        profile_sample = getattr(static, "profileSample", None) if static else None
-        profile_sample_type = getattr(static, "profileSampleType", None) if static else None
+        static = resolve_static_sampling_config(
+            sample_config=sample_config,
+            row_count=self.get_total_row_count(),
+        )
+        profile_sample = static.profileSample if static else None
+        profile_sample_type = static.profileSampleType if static else None
         if profile_sample is None or (profile_sample_type == ProfileSampleType.PERCENTAGE and profile_sample == 100):
             return None, None
         if DatabaseServiceType.Mssql in [
@@ -490,16 +494,19 @@ def calculate_nounce(self, max_nounce=2**32 - 1) -> int:
         config = self.runtime_params.table_profile_config
         profile_sample_config = config.profileSampleConfig if config else None
         sample_config = profile_sample_config.root if profile_sample_config else None
-        static = sample_config.config if sample_config else None
-        profile_sample = getattr(static, "profileSample", 100)
-        profile_sample_type = getattr(static, "profileSampleType", None)
+        row_count = self.get_total_row_count()
+        static = resolve_static_sampling_config(
+            sample_config=sample_config,
+            row_count=row_count,
+        )
+        profile_sample = static.profileSample if static else None
+        profile_sample_type = static.profileSampleType if static else None
         if profile_sample_type == ProfileSampleType.PERCENTAGE:
-            return int(max_nounce * profile_sample / 100)
+            return int(max_nounce * ((profile_sample or 100) / 100))
         if profile_sample_type == ProfileSampleType.ROWS:
-            row_count = self.get_total_row_count()
             if row_count is None:
                 raise ValueError("Row count is required for ROWS profile sample type")
-            return int(max_nounce * (profile_sample / row_count))
+            return int(max_nounce * ((profile_sample or row_count) / row_count))
         raise ValueError("Invalid profile sample type")
 
     def get_row_diff_test_case_result(

ingestion/src/metadata/great_expectations/action.py

@@ -22,7 +22,7 @@
 from typing import Dict, List, Optional, Union, cast  # noqa: UP035
 
 from great_expectations.checkpoint.actions import ValidationAction
-from great_expectations.core import ExpectationConfiguration
+from great_expectations.core import ExpectationConfiguration  # type: ignore
 from great_expectations.core.batch import Batch
 from great_expectations.core.batch_spec import (
     RuntimeDataBatchSpec,
@@ -32,8 +32,8 @@
 from great_expectations.core.expectation_validation_result import (
     ExpectationSuiteValidationResult,
 )
-from great_expectations.data_asset.data_asset import DataAsset
-from great_expectations.data_context.data_context import DataContext
+from great_expectations.data_asset.data_asset import DataAsset  # type: ignore
+from great_expectations.data_context.data_context import DataContext  # type: ignore
 
 from metadata.generated.schema.type.basic import Timestamp
 
@@ -115,7 +115,7 @@ def __init__(
         table_name: Optional[str] = None,  # noqa: UP045
         expectation_suite_table_config_map: Optional[Dict[str, Dict[str, str]]] = None,  # noqa: UP006, UP045
     ):
-        super().__init__(data_context, name=name)
+        super().__init__(data_context, name=name)  # type: ignore
         self.database_service_name = database_service_name
         self.database_name = database_name
         self.table_name = table_name
@@ -153,7 +153,7 @@ def _run(  # pylint: disable=unused-argument
         expectation_suite_name = None
 
         if expectation_suite_identifier:
-            expectation_suite_name = expectation_suite_identifier.expectation_suite_name
+            expectation_suite_name = expectation_suite_identifier.expectation_suite_name  # type: ignore
             self.expectation_suite = self.data_context.get_expectation_suite(expectation_suite_name)
 
         check_point_spec = self._get_checkpoint_batch_spec(data_asset)
@@ -182,7 +182,7 @@ def _run(  # pylint: disable=unused-argument
 
         if table_entity:
             for result in validation_result_suite.results:
-                self._handle_test_case(result, table_entity)
+                self._handle_test_case(result, table_entity)  # type: ignore
 
     @staticmethod
     def _get_checkpoint_batch_spec(

ingestion/src/metadata/great_expectations/action1xx.py

@@ -22,15 +22,15 @@
 from typing import Dict, List, Literal, Optional, Union, cast  # noqa: UP035
 
 from great_expectations.checkpoint import (
-    ActionContext,
-    CheckpointResult,
-    ValidationAction,
+    ActionContext,  # type: ignore
+    CheckpointResult,  # type: ignore
+    ValidationAction,  # type: ignore
 )
 from great_expectations.core.batch import Batch
 from great_expectations.core.expectation_validation_result import (
     ExpectationSuiteValidationResultMeta,
 )
-from great_expectations.datasource.fluent import DataAsset
+from great_expectations.datasource.fluent import DataAsset  # type: ignore
 from great_expectations.validator.validator import Validator
 from sqlalchemy.engine.base import Connection, Engine
 from sqlalchemy.engine.url import URL
@@ -83,7 +83,7 @@ class OpenMetadataValidationAction1xx(ValidationAction):
             Format: {"suite_name": {"database_name": "db", "schema_name": "schema", "table_name": "table"}}
     """
 
-    type: Literal["open_metadata_validation_action"] = "open_metadata_validation_action"
+    type: Literal["open_metadata_validation_action"] = "open_metadata_validation_action"  # type: ignore
     name: str = "OpenMetadataValidationAction"
     config_file_path: Optional[str] = None  # noqa: UP045
     database_service_name: Optional[str] = None  # noqa: UP045
@@ -153,7 +153,7 @@ def run(
 
             if table_entity:
                 for result in v.results:
-                    self._handle_test_case(result, table_entity)
+                    self._handle_test_case(result, table_entity)  # type: ignore
 
     @staticmethod
     def _get_checkpoint_batch_spec(meta: Union[ExpectationSuiteValidationResultMeta, dict]):  # noqa: UP007

ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py

@@ -314,7 +314,7 @@ def __read_manifest(
                     # For remote repositories, clone the dependency as before
                     url_parsed = giturlparse.parse(remote_git_url)
                     _clone_repo(
-                        f"{url_parsed.owner}/{url_parsed.repo}",  # pylint: disable=E1101
+                        f"{url_parsed.owner}/{url_parsed.repo}",  # type: ignore
                         f"{repo.path}/{IMPORTED_PROJECTS_DIR}/{remote_name}",
                         credentials,
                     )

ingestion/src/metadata/ingestion/source/dashboard/powerbi/metadata.py

@@ -675,6 +675,7 @@ def create_report_dashboard_lineage(
     ) -> Iterable[Either[CreateDashboardRequest]]:
         """Create lineage between report and dashboard"""
         try:
+            logger.debug(f"Processing to create report and dashboard lineage for dashboard: {dashboard_details.id}")
             charts = dashboard_details.tiles
             dashboard_fqn = fqn.build(
                 self.metadata,
@@ -686,9 +687,22 @@ def create_report_dashboard_lineage(
                 entity=Dashboard,
                 fqn=dashboard_fqn,
             )
+            if not dashboard_entity:
+                logger.debug(
+                    f"Dashboard entity not found to create lineage between report and dashboard for dashboard: {dashboard_details.id}"
+                )
+                return
             for chart in charts or []:
+                if chart.reportId:
+                    logger.debug(f"Dashboard's chart {chart.id} is linked with report id: {str(chart.reportId)}")  # noqa: RUF010
+                else:
+                    logger.debug(f"Dashboard's chart {chart.id} is not linked with any report")
+                    continue
                 report = self._fetch_report_from_workspace(chart.reportId)
                 if report:
+                    logger.debug(
+                        f"Fetched report details for report id: {str(chart.reportId)} from workspace data to create lineage with dashboard: {dashboard_details.id}"  # noqa: RUF010
+                    )
                     report_fqn = fqn.build(
                         self.metadata,
                         entity_type=Dashboard,
@@ -699,9 +713,19 @@ def create_report_dashboard_lineage(
                         entity=Dashboard,
                         fqn=report_fqn,
                     )
-
+                    if not report_entity:
+                        logger.debug(
+                            f"Report entity not found to create lineage between report={report.id} and dashboard={dashboard_details.id}"
+                        )
                     if report_entity and dashboard_entity:
+                        logger.debug(
+                            f"Creating lineage between report={report.id} and dashboard={dashboard_details.id}"
+                        )
                         yield self._get_add_lineage_request(to_entity=dashboard_entity, from_entity=report_entity)
+                else:
+                    logger.debug(
+                        f"Could not fetch report with report id: {str(chart.reportId)} from workspace data to create lineage with dashboard: {dashboard_details.id}"  # noqa: RUF010
+                    )
         except Exception as exc:  # pylint: disable=broad-except
             yield Either(
                 left=StackTraceError(
@@ -729,6 +753,8 @@ def _get_dataset_ids_from_report_datasources(self, report_id: str) -> List[str]:
                 )
                 if match:
                     dataset_ids.append(match.group(1))
+        if dataset_ids:
+            logger.debug(f"Extracted dataset IDs from report datasources API call for report_id={report_id}")
         return dataset_ids
 
     def create_datamodel_report_lineage(
@@ -740,6 +766,7 @@ def create_datamodel_report_lineage(
         create the lineage between datamodel and report
         """
         try:
+            logger.debug(f"Processing to create datamodel and report lineage for report: {dashboard_details.id}")
             report_fqn = fqn.build(
                 self.metadata,
                 entity_type=Dashboard,
@@ -750,10 +777,19 @@ def create_datamodel_report_lineage(
                 entity=Dashboard,
                 fqn=report_fqn,
             )
+            if not report_entity:
+                logger.debug(
+                    f"Report entity not found to create lineage between datamodel and report for report: {dashboard_details.id}"
+                )
+                return
             dataset_ids = []
             if dashboard_details.datasetId:
+                logger.debug(f"Report linked datasetId is present in api response for report: {dashboard_details.id}")
                 dataset_ids = [dashboard_details.datasetId]
             else:
+                logger.debug(
+                    f"Processing to get report datasources from API to extract datasetIds for report: {dashboard_details.id} as datasetId is not present in api response"
+                )
                 dataset_ids = self._get_dataset_ids_from_report_datasources(report_id=dashboard_details.id)
 
             if dataset_ids:
@@ -768,6 +804,10 @@ def create_datamodel_report_lineage(
                         entity=DashboardDataModel,
                         fqn=datamodel_fqn,
                     )
+                    if not datamodel_entity:
+                        logger.debug(
+                            f"Data model entity not found for dataset_id={str(dataset_id)} while creating lineage with report={str(dashboard_details.id)}"  # noqa: RUF010
+                        )
                     if datamodel_entity and report_entity:
                         logger.debug(
                             f"Creating lineage between datamodel={str(dataset_id)} and report={str(dashboard_details.id)}"  # noqa: RUF010
@@ -778,7 +818,7 @@ def create_datamodel_report_lineage(
                         )
             else:
                 logger.debug(
-                    f"Skipping datamodel and report lineage for {dashboard_details.id} as datasetId is not found"
+                    f"Skipping datamodel and report lineage for report: {dashboard_details.id} as datasetId is not found on api response and also could not be extracted from report datasources API call"
                 )
 
         except Exception as exc:  # pylint: disable=broad-except
@@ -2062,7 +2102,7 @@ def _fetch_dataset_from_workspace(self, dataset_id: Optional[str]) -> Optional[D
                     return dataset_data
         return None
 
-    def _fetch_report_from_workspace(self, report_id: Optional[str]) -> Optional[Dataset]:  # noqa: UP045
+    def _fetch_report_from_workspace(self, report_id: Optional[str]) -> Optional[PowerBIReport]:  # noqa: UP045
         """
         Method to search the report using id in the workspace dict
         """