Merge branch 'main' into update-1.5.11-whats-new-data

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

@@ -1728,12 +1728,11 @@ WHERE JSON_EXTRACT(json, '$.offset') IS NOT NULL
 
 -- Create table successful_sent_change_events for storing successfully sent events per alert
 CREATE TABLE IF NOT EXISTS successful_sent_change_events (
-    id VARCHAR(36) NOT NULL,
     change_event_id VARCHAR(36) NOT NULL,
     event_subscription_id VARCHAR(36) NOT NULL,
     json JSON NOT NULL,
     timestamp BIGINT UNSIGNED NOT NULL,
-    PRIMARY KEY (id)
+    PRIMARY KEY (change_event_id, event_subscription_id)
 );
 
 -- Create an index on the event_subscription_id column in the successful_sent_change_events table

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

@@ -1715,11 +1715,11 @@ WHERE json -> 'offset' IS NOT NULL
 
 -- Create table successful_sent_change_events for storing successfully sent events per alert
 CREATE TABLE IF NOT EXISTS successful_sent_change_events (
-    id VARCHAR(36) PRIMARY KEY,
     change_event_id VARCHAR(36) NOT NULL,
     event_subscription_id VARCHAR(36) NOT NULL,
     json jsonb NOT NULL,
-    timestamp BIGINT NOT NULL
+    timestamp BIGINT NOT NULL,
+    PRIMARY KEY (change_event_id, event_subscription_id)
 );
 
 -- Create an index on the event_subscription_id column in the successful_sent_change_events table

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

@@ -54,7 +54,7 @@ def name_search_query_es(entity: Type[T]) -> str:
         but looked as `Random User`, we want to find this match.
         """
         return (
-            "/search/query?q={name}&from={from_}&size={size}&index="
+            "/search/query?q={name} AND isBot:false&from={from_}&size={size}&index="
             + ES_INDEX_MAP[entity.__name__]
         )
 

ingestion/src/metadata/ingestion/source/database/databricks/metadata.py

@@ -112,6 +112,12 @@ class MAP(String):
     }
 )
 
+
+def format_schema_name(schema):
+    # Adds back quotes(``) if hyphen(-) in schema name
+    return f"`{schema}`" if "-" in schema else schema
+
+
 # This method is from hive dialect originally but
 # is overridden to optimize DESCRIBE query execution
 def _get_table_columns(self, connection, table_name, schema, db_name):
@@ -154,6 +160,7 @@ def _get_table_columns(self, connection, table_name, schema, db_name):
 
 def _get_column_rows(self, connection, table_name, schema, db_name):
     # get columns and strip whitespace
+    schema = format_schema_name(schema=schema)
     table_columns = _get_table_columns(  # pylint: disable=protected-access
         self, connection, table_name, schema, db_name
     )
@@ -388,6 +395,7 @@ def get_table_type(self, connection, database, schema, table):
                 database_name=database, schema_name=schema, table_name=table
             )
         else:
+            schema = format_schema_name(schema=schema)
             query = f"DESCRIBE TABLE EXTENDED {schema}.{table}"
         rows = get_table_comment_result(
             self,
@@ -755,6 +763,7 @@ def get_table_description(
     ) -> str:
         description = None
         try:
+            schema_name = format_schema_name(schema=schema_name)
             query = DATABRICKS_GET_TABLE_COMMENTS.format(
                 database_name=self.context.get().database,
                 schema_name=schema_name,
@@ -809,7 +818,9 @@ def get_owner_ref(self, table_name: str) -> Optional[EntityReferenceList]:
         try:
             query = DATABRICKS_GET_TABLE_COMMENTS.format(
                 database_name=self.context.get().database,
-                schema_name=self.context.get().database_schema,
+                schema_name=format_schema_name(
+                    schema=self.context.get().database_schema
+                ),
                 table_name=table_name,
             )
             result = self.inspector.dialect.get_table_comment_result(

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

@@ -17,7 +17,7 @@
 from enum import Enum
 from typing import Dict, Iterable, List, Optional, cast
 
-from airflow.models import BaseOperator, DagRun, TaskInstance
+from airflow.models import BaseOperator, DagRun, DagTag, TaskInstance
 from airflow.models.dag import DagModel
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.serialization.serialized_objects import SerializedDAG
@@ -57,6 +57,7 @@
 from metadata.ingestion.api.models import Either
 from metadata.ingestion.api.steps import InvalidSourceException
 from metadata.ingestion.connections.session import create_and_bind_session
+from metadata.ingestion.models.ometa_classification import OMetaTagAndClassification
 from metadata.ingestion.models.pipeline_status import OMetaPipelineStatus
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.pipeline.airflow.lineage_parser import (
@@ -74,9 +75,12 @@
 from metadata.utils.constants import ENTITY_REFERENCE_TYPE_MAP
 from metadata.utils.helpers import clean_uri, datetime_to_ts
 from metadata.utils.logger import ingestion_logger
+from metadata.utils.tag_utils import get_ometa_tag_and_classification, get_tag_labels
 
 logger = ingestion_logger()
 
+AIRFLOW_TAG_CATEGORY = "AirflowTags"
+
 
 class AirflowTaskStatus(Enum):
     SUCCESS = "success"
@@ -152,6 +156,31 @@ def _extract_serialized_task(task: Dict) -> Dict:
             return task["__var"]
         return task
 
+    def get_all_tags(self, dag_id: str) -> List[str]:
+        try:
+            tag_query = (
+                self.session.query(DagTag.name)
+                .filter(DagTag.dag_id == dag_id)
+                .distinct()
+                .all()
+            )
+            return [tag[0] for tag in tag_query]
+        except Exception as exc:
+            logger.debug(traceback.format_exc())
+            logger.warning(f"Could not extract tags details due to {exc}")
+        return []
+
+    def yield_tag(
+        self, pipeline_details: AirflowDagDetails
+    ) -> Iterable[Either[OMetaTagAndClassification]]:
+        yield from get_ometa_tag_and_classification(
+            tags=self.get_all_tags(dag_id=pipeline_details.dag_id),
+            classification_name=AIRFLOW_TAG_CATEGORY,
+            tag_description="Airflow Tag",
+            classification_description="Tags associated with airflow entities.",
+            include_tags=self.source_config.includeTags,
+        )
+
     def get_pipeline_status(self, dag_id: str) -> List[DagRun]:
         """
         Return the DagRuns of given dag
@@ -441,7 +470,6 @@ def yield_pipeline(
         try:
             # Airflow uses /dags/dag_id/grid to show pipeline / dag
             source_url = f"{clean_uri(self.service_connection.hostPort)}/dags/{pipeline_details.dag_id}/grid"
-
             pipeline_request = CreatePipelineRequest(
                 name=EntityName(pipeline_details.dag_id),
                 description=Markdown(pipeline_details.description)
@@ -459,6 +487,12 @@ def yield_pipeline(
                 service=FullyQualifiedEntityName(self.context.get().pipeline_service),
                 owners=self.get_owner(pipeline_details.owner),
                 scheduleInterval=pipeline_details.schedule_interval,
+                tags=get_tag_labels(
+                    metadata=self.metadata,
+                    tags=pipeline_details.data.dag.tags,
+                    classification_name=AIRFLOW_TAG_CATEGORY,
+                    include_tags=self.source_config.includeTags,
+                ),
             )
             yield Either(right=pipeline_request)
             self.register_record(pipeline_request=pipeline_request)

ingestion/src/metadata/ingestion/source/pipeline/dagster/metadata.py

@@ -161,7 +161,9 @@ def yield_pipeline(
                 )
             )
 
-    def yield_tag(self, *_, **__) -> Iterable[Either[OMetaTagAndClassification]]:
+    def yield_tag(
+        self, pipeline_details: DagsterPipeline
+    ) -> Iterable[Either[OMetaTagAndClassification]]:
         yield from get_ometa_tag_and_classification(
             tags=[self.context.get().repository_name],
             classification_name=DAGSTER_TAG_CATEGORY,

ingestion/src/metadata/ingestion/source/pipeline/pipeline_service.py

@@ -222,7 +222,9 @@ def _get_table_fqn_from_om(self, table_details: TableDetails) -> Optional[str]:
             f"Table FQN not found for table: {table_details} within services: {services}"
         )
 
-    def yield_tag(self, *args, **kwargs) -> Iterable[Either[OMetaTagAndClassification]]:
+    def yield_tag(
+        self, pipeline_details: Any
+    ) -> Iterable[Either[OMetaTagAndClassification]]:
         """Method to fetch pipeline tags"""
 
     def close(self):

ingestion/tests/cli_e2e/test_cli_quicksight.py

@@ -14,6 +14,8 @@
 """
 from typing import List
 
+from metadata.ingestion.api.status import Status
+
 from .common.test_cli_dashboard import CliCommonDashboard
 
 
@@ -50,7 +52,7 @@ def expected_filtered_mix(self) -> int:
         return 2
 
     def expected_filtered_sink_mix(self) -> int:
-        return 2
+        return 4
 
     # Quicksight do not ingest tags
     def expected_tags(self) -> int:
@@ -71,4 +73,34 @@ def expected_datamodels(self) -> int:
         return 0
 
     def expected_dashboards_and_charts_after_patch(self) -> int:
-        return 0
+        return 7
+
+    def assert_for_vanilla_ingestion(
+        self, source_status: Status, sink_status: Status
+    ) -> None:
+        """
+        We are overriding this method because of diff.
+        of 1 in source and sink records
+        """
+        self.assertTrue(len(source_status.failures) == 0)
+        self.assertTrue(len(source_status.warnings) == 0)
+        self.assertTrue(len(source_status.filtered) == 0)
+        self.assertEqual(
+            (len(source_status.records) + len(source_status.updated_records)),
+            self.expected_dashboards_and_charts_after_patch()
+            + self.expected_tags()
+            + self.expected_lineage()
+            + self.expected_datamodels()
+            + self.expected_datamodel_lineage(),
+        )
+        self.assertTrue(len(sink_status.failures) == 0)
+        self.assertTrue(len(sink_status.warnings) == 0)
+        # We are getting here diff of 1 element in case of the service ingested.
+        self.assertTrue(
+            (len(sink_status.records) + len(sink_status.updated_records))
+            <= self.expected_dashboards_and_charts_after_patch()
+            + self.expected_tags()
+            + self.expected_lineage()
+            + self.expected_datamodels()
+            + self.expected_datamodel_lineage(),
+        )

ingestion/tests/integration/ometa/test_ometa_user_api.py

@@ -161,6 +161,11 @@ def test_es_search_from_name(self):
         # Non existing email returns None
         self.assertIsNone(self.metadata.get_reference_by_name(name="idonotexist"))
 
+        # when searching for "data" user we should not get DataInsightsApplicationBot in result
+        team_data = self.metadata.get_reference_by_name(name="data").root[0]
+        self.assertEqual(team_data.name, "Data")
+        self.assertEqual(team_data.type, "team")
+
         # We can get the user matching its name
         self.assertEqual(
             self.user_1.id,

ingestion/tests/unit/test_databricks_schema.py

@@ -0,0 +1,21 @@
+import pytest
+
+from metadata.ingestion.source.database.databricks.metadata import format_schema_name
+
+
+@pytest.mark.parametrize(
+    "input_schema, expected_schema",
+    [
+        ("test_schema-name", "`test_schema-name`"),
+        ("test_schema_name", "test_schema_name"),
+        ("schema-with-hyphen", "`schema-with-hyphen`"),
+        ("schema_with_underscore", "schema_with_underscore"),
+        ("validSchema", "validSchema"),
+    ],
+)
+def test_schema_name_sanitization(input_schema, expected_schema):
+    """
+    Test sanitization of schema names by adding backticks only around hyphenated names.
+    """
+    sanitized_schema = format_schema_name(input_schema)
+    assert sanitized_schema == expected_schema

openmetadata-docs/content/v1.5.x/how-to-guides/admin-guide/how-to-add-custom-logo.md

@@ -7,16 +7,21 @@ slug: /how-to-guides/admin-guide/how-to-add-custom-logo
 
 OpenMetadata allows you to customize the logo and favicons. In this guide we will see how to add custom logo and favicon.
 
-1. Navigate to **Setting >> OpenMetadata >> Custom Logo**
+1. Navigate to **Setting >> Preferences >> Themes**
 
-page and then scroll down to OpenMetadata section and navigate to `Custom Logo` settings and click on the `Edit` button.
+page and then you will get option to enter `Custom Logo` url.
 
 {% image
 src="/images/v1.5/how-to-guides/custom-logo/logo.png"
+alt="Custom Logo"
+/%}
+
+{% image
+src="/images/v1.5/how-to-guides/custom-logo/logo0.1.png"
 alt="Custom Logo Setting"
 /%}
 
-2. Click on the **Edit**. Enter the values in the edit form and **Save** the changes.
+2. Enter the values in the edit form and **Save** the changes.
 
 {% image
 src="/images/v1.5/how-to-guides/custom-logo/custom-logo-form.png"

openmetadata-docs/content/v1.6.x-SNAPSHOT/how-to-guides/admin-guide/how-to-add-custom-logo.md

@@ -7,30 +7,35 @@ slug: /how-to-guides/admin-guide/how-to-add-custom-logo
 
 OpenMetadata allows you to customize the logo and favicons. In this guide we will see how to add custom logo and favicon.
 
-1. Navigate to **Setting >> OpenMetadata >> Custom Logo**
+1. Navigate to **Setting >> Preferences >> Themes**
 
-page and then scroll down to OpenMetadata section and navigate to `Custom Logo` settings and click on the `Edit` button.
+page and page and then you will get option to enter `Custom Logo` url.
 
 {% image
-src="/images/v1.5/how-to-guides/custom-logo/logo.png"
+src="/images/v1.6/how-to-guides/custom-logo/logo.png"
+alt="Custom Logo"
+/%}
+
+{% image
+src="/images/v1.6/how-to-guides/custom-logo/logo0.1.png"
 alt="Custom Logo Setting"
 /%}
 
-2. Click on the **Edit**. Enter the values in the edit form and **Save** the changes.
+2. Enter the values in the edit form and **Save** the changes.
 
 {% image
-src="/images/v1.5/how-to-guides/custom-logo/custom-logo-form.png"
+src="/images/v1.6/how-to-guides/custom-logo/custom-logo-form.png"
 alt="custom-logo-form"
 /%}
 
 3. After saving the changes, refresh the page to view the updated logos and favicon.
 
 {% image
-src="/images/v1.5/how-to-guides/custom-logo/custom-logo-application.png"
+src="/images/v1.6/how-to-guides/custom-logo/custom-logo-application.png"
 alt="custom-logo-application"
 /%}
 
 {% image
-src="/images/v1.5/how-to-guides/custom-logo/custom-logo-login-page.png"
+src="/images/v1.6/how-to-guides/custom-logo/custom-logo-login-page.png"
 alt="custom-logo-login-page"
 /%}

openmetadata-service/pom.xml

@@ -16,7 +16,7 @@
     <sonar.coverage.jacoco.xmlReportPaths>${project.basedir}/target/site/jacoco-aggregate/jacoco.xml</sonar.coverage.jacoco.xmlReportPaths>
     <sonar.tests>${project.basedir}/src/test/java</sonar.tests>
     <org.testcontainers.version>1.20.3</org.testcontainers.version>
-    <awssdk.version>2.29.1</awssdk.version>
+    <awssdk.version>2.29.11</awssdk.version>
     <azure-identity.version>1.14.0</azure-identity.version>
     <azure-kv.version>4.9.0</azure-kv.version>
     <azure-identity-extensions.version>1.0.0</azure-identity-extensions.version>

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java

@@ -116,12 +116,11 @@ public void handleFailedEvent(EventPublisherException ex, boolean errorOnSub) {
   private void recordSuccessfulChangeEvent(UUID eventSubscriptionId, ChangeEvent event) {
     Entity.getCollectionDAO()
         .eventSubscriptionDAO()
-        .insertSuccessfulChangeEvent(
-            UUID.randomUUID().toString(),
+        .upsertSuccessfulChangeEvent(
             event.getId().toString(),
             eventSubscriptionId.toString(),
             JsonUtils.pojoToJson(event),
-            event.getTimestamp());
+            System.currentTimeMillis());
   }
 
   private EventSubscriptionOffset loadInitialOffset(JobExecutionContext context) {