ISSUE #3032 (#27912)

* feat: move flat sampling to sampling config + dynamic sampling option

* feat: move flat sampling on the backend to sample profile conifg object

* feat: fix circular import

* feat: align UI with new profiler config

* feat: fix json schema

* feat: align python imports with new schema path

* feat: update migration to look at extension

* feat: remove enable

* feat: remove enable

* feat: added titles to sample config

* feat: generated ts classes

* feat: addressed comments

* feat: change sample config instantiation to match new structure

* feat: removed backward compatible fields

* feat: ran java linting

* feat: updated imports to point to generated files

* feat: added dynamic sampler resolution logic

* feat: ran python linting

* feat: remove duplicate migration

* chore: merge upstream and clean conflicts

* feat: update logic to support dynamic and static sampling

* feat: adjust sample config call

* feat: test for statis, dynamic, row count and tier methods

* feat: more sample config unit tests

* feat: added tests for metric and sampling

* feat: added tests to validate fallback is not called i nmetric computers

* feat: strengthen profiler validation tests

* feat: fix sampling config

* feat: fix sampling config

* feat: fix sampling config

* feat: generated typescript models

* feat: fixed missing dq pipeline migration

* feat: fixed static check

* feat: fixed ci failures

* feat: fixed ci failures

* feat: fixed unit tests faioure and linting

* feat: fixed integration tests failures

* chore: fixe burstiq refactor

* chore: fix trino ci failures

* chore: revert baseline.json file

* chore: fix sampler availabl burst iq changes

* feat: added smart sampling radio button

* feat: ignore static checks errors

* feat: ran ts linting

* feat burstiq infinite recursion issue with dynamic as default

* feat: translate i8n keys

* feat: fix failing tests

Author: TeddyCr

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

@@ -422,6 +422,8 @@
     # TODO: Remove once no unit test requires testcontainers
     "testcontainers",
     VERSIONS["factory-boy"],
+    *plugins["exasol"],
+    *plugins["teradata"],
 }
 
 test = {
@@ -490,6 +492,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/database/presto/service_spec.py

@@ -1,4 +1,6 @@
 from metadata.ingestion.source.database.presto.metadata import PrestoSource
 from metadata.utils.service_spec.default import DefaultDatabaseSpec
 
-ServiceSpec = DefaultDatabaseSpec(metadata_source_class=PrestoSource)
+ServiceSpec = DefaultDatabaseSpec(
+    metadata_source_class=PrestoSource,  # type: ignore
+)