Merge branch 'main' into feat/issue-12787-hive-metastore-mssql-oracle

Author: SaaiAravindhRaja

.gitignore

@@ -192,3 +192,4 @@ ingestion/.claude/agents
 # AI scaffold working documents — stay local, never committed
 **/CONNECTOR_CONTEXT.md
 
+test-results/

ingestion/src/metadata/ingestion/ometa/mixins/table_mixin.py

@@ -260,6 +260,20 @@ def get_sample_data(self, table: Table) -> Optional[Table]:
 
         return None
 
+    def delete_sample_data(self, table: Table) -> None:
+        """
+        DELETE call for the /sampleData endpoint for a given Table
+        """
+        try:
+            self.client.delete(
+                f"{self.get_suffix(Table)}/{table.id.root}/sampleData",
+            )
+        except Exception as exc:
+            logger.debug(traceback.format_exc())
+            logger.warning(
+                f"Error trying to DELETE sample data for {table.fullyQualifiedName.root}: {exc}"
+            )
+
     def add_pipeline_observability(
         self, table_id: Uuid, pipeline_observability: List[PipelineObservability]
     ) -> Optional[Table]:

ingestion/src/metadata/ingestion/source/api/rest/connection.py

@@ -45,6 +45,7 @@
     validate_openapi_schema,
 )
 from metadata.utils.constants import THREE_MIN
+from metadata.utils.ssl_registry import get_verify_ssl_fn
 
 
 class SchemaURLError(Exception):
@@ -67,10 +68,16 @@ def get_connection(connection: RestConnection) -> Union[Response, Dict]:
     """
     schema_conn = connection.openAPISchemaConnection
     if isinstance(schema_conn, OpenAPISchemaURL):
+        verify_ssl_fn = get_verify_ssl_fn(connection.verifySSL)
+        verify = verify_ssl_fn(connection.sslConfig)
+        if verify is None:
+            verify = True
+        headers = {}
         if connection.token:
-            headers = {"Authorization": f"Bearer {connection.token.get_secret_value()}"}
-            return requests.get(schema_conn.openAPISchemaURL, headers=headers)
-        return requests.get(schema_conn.openAPISchemaURL)
+            headers["Authorization"] = f"Bearer {connection.token.get_secret_value()}"
+        return requests.get(
+            schema_conn.openAPISchemaURL, headers=headers, verify=verify
+        )
 
     if isinstance(schema_conn, OpenAPISchemaFilePath):
         return parse_openapi_schema_from_file(schema_conn.openAPISchemaFilePath)

ingestion/src/metadata/ingestion/source/messaging/common_broker_source.py

@@ -81,6 +81,11 @@ def __init__(
     ):
         super().__init__(config, metadata)
         self.generate_sample_data = self.config.sourceConfig.config.generateSampleData
+        if (
+            self.generate_sample_data
+            and self._is_sample_data_storing_globally_disabled()
+        ):
+            self.generate_sample_data = False
         self.service_connection = self.config.serviceConnection.root.config
         self.admin_client = self.connection.admin_client
         self.schema_registry_client = self.connection.schema_registry_client

ingestion/src/metadata/ingestion/source/messaging/kinesis/metadata.py

@@ -73,6 +73,11 @@ def __init__(
     ):
         super().__init__(config, metadata)
         self.generate_sample_data = self.config.sourceConfig.config.generateSampleData
+        if (
+            self.generate_sample_data
+            and self._is_sample_data_storing_globally_disabled()
+        ):
+            self.generate_sample_data = False
         self.kinesis = self.connection
 
     @classmethod

ingestion/src/metadata/ingestion/source/messaging/messaging_service.py

@@ -13,13 +13,17 @@
 """
 
 from abc import ABC, abstractmethod
-from typing import Any, Iterable, List, Optional, Set
+from typing import Any, Iterable, List, Optional, Set, cast
 
 from pydantic import BaseModel, Field
 from typing_extensions import Annotated
 
 from metadata.generated.schema.api.data.createTopic import CreateTopicRequest
 from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest
+from metadata.generated.schema.configuration.profilerConfiguration import (
+    ProfilerConfiguration,
+    SampleDataIngestionConfig,
+)
 from metadata.generated.schema.entity.data.topic import Topic, TopicSampleData
 from metadata.generated.schema.entity.services.messagingService import (
     MessagingConnection,
@@ -153,6 +157,32 @@ def __init__(
     def name(self) -> str:
         return self.service_connection.type.name
 
+    def _is_sample_data_storing_globally_disabled(self) -> bool:
+        """Check if storing sample data is globally disabled via profiler configuration.
+
+        Returns True if storing of sample data is disabled in the global
+        profiler config, meaning sample data should not be fetched for
+        messaging sources.
+        """
+        try:
+            settings = self.metadata.get_profiler_config_settings()
+            if not settings or not settings.config_value:
+                return False
+            profiler_config = cast(ProfilerConfiguration, settings.config_value)
+            sample_data_config = profiler_config.sampleDataConfig
+            if sample_data_config is None:
+                return False
+            sample_data_config = cast(SampleDataIngestionConfig, sample_data_config)
+            if not sample_data_config.storeSampleData:
+                logger.info(
+                    "Global profiler configuration disables storing "
+                    "of sample data. Overriding source configuration."
+                )
+                return True
+        except Exception as exc:
+            logger.debug(f"Could not fetch global profiler config: {exc}")
+        return False
+
     @abstractmethod
     def yield_topic(self, topic_details: Any) -> Iterable[Either[CreateTopicRequest]]:
         """

ingestion/src/metadata/ingestion/source/messaging/pubsub/metadata.py

@@ -75,6 +75,11 @@ def __init__(
     ):
         super().__init__(config, metadata)
         self.generate_sample_data = self.config.sourceConfig.config.generateSampleData
+        if (
+            self.generate_sample_data
+            and self._is_sample_data_storing_globally_disabled()
+        ):
+            self.generate_sample_data = False
         self.pubsub = self.connection
         self.project_id = self.pubsub.project_id
 

ingestion/src/metadata/ingestion/source/pipeline/airflow/api/source.py

@@ -42,6 +42,7 @@
     SourceUrl,
     Timestamp,
 )
+from metadata.generated.schema.type.entityReferenceList import EntityReferenceList
 from metadata.ingestion.api.models import Either
 from metadata.ingestion.api.steps import InvalidSourceException
 from metadata.ingestion.models.ometa_classification import OMetaTagAndClassification
@@ -126,6 +127,23 @@ def _get_dag_source_url(self, dag_id: str) -> str:
             return f"{host}/dags/{quote(dag_id)}"
         return f"{host}/dags/{quote(dag_id)}/grid"
 
+    def get_owners(self, owners: Optional[List[str]]) -> Optional[EntityReferenceList]:
+        if not self.source_config.includeOwners or not owners:
+            return None
+        refs = EntityReferenceList(root=[])
+        for owner_name in owners:
+            try:
+                ref = self.metadata.get_reference_by_name(
+                    name=owner_name, is_owner=True
+                )
+                if ref:
+                    refs.root.extend(ref.root)
+            except Exception as exc:
+                logger.warning(
+                    f"Error while getting details of user {owner_name} - {exc}"
+                )
+        return refs if refs.root else None
+
     def _build_tasks(self, dag_details: AirflowApiDagDetails) -> List[Task]:
         return [
             Task(
@@ -164,6 +182,7 @@ def yield_pipeline(
                 ),
                 tasks=self._build_tasks(pipeline_details),
                 service=FullyQualifiedEntityName(self.context.get().pipeline_service),
+                owners=self.get_owners(pipeline_details.owners),
                 scheduleInterval=pipeline_details.schedule_interval,
                 tags=get_tag_labels(
                     metadata=self.metadata,

ingestion/src/metadata/sampler/processor.py

@@ -15,6 +15,9 @@
 from copy import deepcopy
 from typing import Optional, Type, cast
 
+from metadata.generated.schema.configuration.profilerConfiguration import (
+    ProfilerConfiguration,
+)
 from metadata.generated.schema.entity.data.database import Database
 from metadata.generated.schema.entity.data.table import Table
 from metadata.generated.schema.entity.services.connections.database.bigQueryConnection import (
@@ -135,9 +138,28 @@ def _run(self, record: ProfilerSourceAndEntity) -> Either[SamplerResponse]:
                 default_sample_config=SampleConfig(),
                 default_sample_data_count=self.source_config.sampleDataCount,
             )
+
+            settings = self.metadata.get_profiler_config_settings()
+            profiler_global_config = (
+                cast(ProfilerConfiguration, settings.config_value) if settings else None
+            )
+
+            sample_data_config = (
+                profiler_global_config.sampleDataConfig
+                if profiler_global_config
+                else None
+            )
+
             sample_data = SampleData(
-                data=sampler_interface.generate_sample_data(),
-                store=self.source_config.storeSampleData,
+                data=sampler_interface.generate_sample_data(
+                    sample_data_config if sample_data_config else None
+                ),
+                store=bool(
+                    self.source_config.storeSampleData
+                    and (
+                        sample_data_config is None or sample_data_config.storeSampleData
+                    )
+                ),
             )
             sampler_interface.close()
             return Either(

ingestion/src/metadata/sampler/sampler_interface.py

@@ -15,6 +15,9 @@
 from abc import ABC, abstractmethod
 from typing import Any, List, Optional, Set, Union
 
+from metadata.generated.schema.configuration.profilerConfiguration import (
+    SampleDataIngestionConfig,
+)
 from metadata.generated.schema.entity.data.database import Database
 from metadata.generated.schema.entity.data.databaseSchema import DatabaseSchema
 from metadata.generated.schema.entity.data.table import (
@@ -240,33 +243,57 @@ def _truncate_cell(value: Any) -> Any:
         return value
 
     @calculate_execution_time(store=False)
-    def generate_sample_data(self) -> Optional[TableData]:
+    def generate_sample_data(
+        self, sample_data_config: Optional[SampleDataIngestionConfig] = None
+    ) -> TableData:
         """Fetch and ingest sample data
 
         Returns:
             TableData: sample data
         """
-        try:
-            logger.debug(
-                f"Fetching sample data for {self.entity.fullyQualifiedName.root}..."
+        if sample_data_config is None:
+            # if there is no global config, default to storing and reading sample data to ensure backward compatibility
+            # and availability of sample data for downstream steps
+            sample_data_config = SampleDataIngestionConfig(
+                storeSampleData=True, readSampleData=True
             )
-            table_data = self.fetch_sample_data(self.columns)
-            # Truncate large cell values to prevent OOM in downstream
-            # processing (NLP, serialization, etc.)
-            table_data.rows = [
-                [self._truncate_cell(cell) for cell in row]
-                for row in table_data.rows[
-                    : min(SAMPLE_DATA_DEFAULT_COUNT, self.sample_limit)
-                ]
-            ]
-            # Only store the data if configured to do so
-            if self.storage_config:
-                upload_sample_data(
-                    data=table_data,
-                    entity=self.entity,
-                    sample_storage_config=self.storage_config,
+
+        if (
+            not sample_data_config.storeSampleData
+            and not sample_data_config.readSampleData
+        ):
+            logger.info(
+                "Both storing and reading of sample data are disabled. Skipping sample data generation."
+            )
+            return TableData(rows=[], columns=[])
+        try:
+
+            # Stores overwrites reading since if we are storing the data, we want to fetch it
+            # as well to pass down the pipeline. If we are not storing, but reading is enabled,
+            # we still want to fetch the data to pass down the pipeline, but we won't store it.
+            if sample_data_config.readSampleData or sample_data_config.storeSampleData:
+                logger.debug(
+                    f"Fetching sample data for {self.entity.fullyQualifiedName.root}..."
                 )
-            return table_data
+                table_data = self.fetch_sample_data(self.columns)
+                # Truncate large cell values to prevent OOM in downstream
+                # processing (NLP, serialization, etc.)
+                table_data.rows = [
+                    [self._truncate_cell(cell) for cell in row]
+                    for row in table_data.rows[
+                        : min(SAMPLE_DATA_DEFAULT_COUNT, self.sample_limit)
+                    ]
+                ]
+                # Only store the data if configured to do so
+                if self.storage_config and sample_data_config.storeSampleData:
+                    upload_sample_data(
+                        data=table_data,
+                        entity=self.entity,
+                        sample_storage_config=self.storage_config,
+                    )
+                return table_data
+
+            return TableData(rows=[], columns=[])
 
         except Exception as err:
             logger.debug(traceback.format_exc())