Fix #14786: Suggestions API

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

@@ -169,5 +169,17 @@ DELETE FROM event_subscription_entity;
 DELETE FROM change_event_consumers;
 DELETE FROM consumers_dlq;
 
-UPDATE ingestion_pipeline_entity SET json = JSON_SET(json, '$.provider', 'user') 
+CREATE TABLE IF NOT EXISTS suggestions (
+ id VARCHAR(36) GENERATED ALWAYS AS (json ->> '$.id') STORED NOT NULL,
+ fqnHash VARCHAR(256) NOT NULL COLLATE ascii_bin,
+ entityLink VARCHAR(256) GENERATED ALWAYS AS (json ->> '$.entityLink') NOT NULL,
+ suggestionType VARCHAR(36) GENERATED ALWAYS AS (json_unquote(json ->> '$.type')) NOT NULL,
+ json JSON NOT NULL,
+ updatedAt BIGINT UNSIGNED GENERATED ALWAYS AS (json ->> '$.updatedAt') NOT NULL,
+ updatedBy VARCHAR(256) GENERATED ALWAYS AS (json ->> '$.updatedBy') NOT NULL,
+ status VARCHAR(256) GENERATED ALWAYS AS (json_unquote(json -> '$.status')) NOT NULL,
+ PRIMARY KEY (id)
+);
+
+UPDATE ingestion_pipeline_entity SET json = JSON_SET(json, '$.provider', 'user')
 WHERE JSON_EXTRACT(json, '$.name') = 'OpenMetadata_dataInsight';

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

@@ -181,5 +181,17 @@ DELETE FROM event_subscription_entity;
 DELETE FROM change_event_consumers;
 DELETE FROM consumers_dlq;
 
+CREATE TABLE IF NOT EXISTS suggestions (
+ id VARCHAR(36) GENERATED ALWAYS AS (json ->> 'id') STORED NOT NULL,
+ fqnHash VARCHAR(256) NOT NULL,
+ entityLink VARCHAR(256) GENERATED ALWAYS AS (json ->> 'entityLink') STORED NOT NULL,
+ suggestionType VARCHAR(36) GENERATED ALWAYS AS (json ->> 'type') STORED NOT NULL,
+ json JSON NOT NULL,
+ updatedAt BIGINT GENERATED ALWAYS AS ((json ->> 'updatedAt')::bigint) STORED NOT NULL,
+ updatedBy VARCHAR(256) GENERATED ALWAYS AS (json ->> 'updatedBy') STORED NOT NULL,
+ status VARCHAR(256) GENERATED ALWAYS AS (json ->> 'status') STORED NOT NULL,
+ PRIMARY KEY (id)
+);
+
 UPDATE ingestion_pipeline_entity SET json = JSONB_SET(json::jsonb, '{provider}', '"user"', true)
 WHERE json->>'name' = 'OpenMetadata_dataInsight';

conf/openmetadata.yaml

@@ -369,4 +369,3 @@ web:
  permission-policy:
  enabled: ${WEB_CONF_PERMISSION_POLICY_ENABLED:-false}
  option: ${WEB_CONF_PERMISSION_POLICY_OPTION:-""}
-

ingestion/src/metadata/applications/auto_tagger.py

@@ -21,6 +21,9 @@
 from metadata.generated.schema.entity.services.ingestionPipelines.status import (
  StackTraceError,
 )
+from metadata.generated.schema.metadataIngestion.application import (
+ OpenMetadataApplicationConfig,
+)
 from metadata.generated.schema.type.tagLabel import (
  LabelType,
  State,
@@ -60,16 +63,18 @@ class AutoTaggerApp(AppRunner):
  jwtToken: "..."
  """
 
- def __init__(self, config: AutoTaggerAppConfig, metadata: OpenMetadata):
+ def __init__(self, config: OpenMetadataApplicationConfig, metadata: OpenMetadata):
  super().__init__(config, metadata)
 
- if not isinstance(config, AutoTaggerAppConfig):
+ if not isinstance(self.app_config, AutoTaggerAppConfig):
  raise InvalidAppConfiguration(
  f"AutoTagger Runner expects an AutoTaggerAppConfig, we got [{config}]"
  )
 
  self._ner_scanner = None
- self.confidence_threshold = config.confidenceLevel or DEFAULT_CONFIDENCE
+ self.confidence_threshold = (
+ self.app_config.confidenceLevel or DEFAULT_CONFIDENCE
+ )
 
  @property
  def name(self) -> str:

ingestion/src/metadata/data_quality/processor/test_case_runner.py

@@ -28,6 +28,7 @@
  test_suite_source_factory,
 )
 from metadata.generated.schema.api.tests.createTestCase import CreateTestCaseRequest
+from metadata.generated.schema.entity.data.table import Table
 from metadata.generated.schema.entity.services.ingestionPipelines.status import (
  StackTraceError,
 )
@@ -200,8 +201,9 @@ def compare_and_create_test_cases(
  ),
  entityLink=EntityLink(
  __root__=entity_link.get_entity_link(
- table_fqn,
- test_case_to_create.columnName,
+ Table,
+ fqn=table_fqn,
+ column_name=test_case_to_create.columnName,
  )
  ),
  testSuite=test_suite_fqn,
@@ -252,8 +254,9 @@ def _update_test_cases(
  updated_test_case = self.metadata.patch_test_case_definition(
  source=test_case,
  entity_link=entity_link.get_entity_link(
- table_fqn,
- test_case_definition.columnName,
+ Table,
+ fqn=table_fqn,
+ column_name=test_case_definition.columnName,
  ),
  test_case_parameter_values=test_case_definition.parameterValues,
  )

ingestion/src/metadata/great_expectations/action.py

@@ -420,8 +420,9 @@ def _handle_test_case(
  test_case = self.ometa_conn.get_or_create_test_case(
  test_case_fqn,
  entity_link=get_entity_link(
- table_entity.fullyQualifiedName.__root__,
- fqn.split_test_case_fqn(test_case_fqn).column,
+ Table,
+ fqn=table_entity.fullyQualifiedName.__root__,
+ column_name=fqn.split_test_case_fqn(test_case_fqn).column,
  ),
  test_suite_fqn=test_suite.fullyQualifiedName.__root__,
  test_definition_fqn=test_definition.fullyQualifiedName.__root__,

ingestion/src/metadata/ingestion/api/status.py

@@ -56,6 +56,10 @@ def scanned(self, record: Any) -> None:
  else:
  self.records.append(log_name)
 
+ def updated(self, record: Any) -> None:
+ if log_name := get_log_name(record):
+ self.updated_records.append(log_name)
+
  def warning(self, key: str, reason: str) -> None:
  self.warnings.append({key: reason})
 

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

@@ -0,0 +1,41 @@
+# Copyright 2021 Collate
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+# http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+Mixin class containing Suggestions specific methods
+
+To be used by OpenMetadata class
+"""
+from metadata.generated.schema.entity.feed.suggestion import Suggestion
+from metadata.ingestion.ometa.client import REST
+from metadata.utils.logger import ometa_logger
+
+logger = ometa_logger()
+
+
+class OMetaSuggestionsMixin:
+ """
+ OpenMetadata API methods related to the Suggestion Entity
+
+ To be inherited by OpenMetadata
+ """
+
+ client: REST
+
+ def update_suggestion(self, suggestion: Suggestion) -> Suggestion:
+ """
+ Update an existing Suggestion with new fields
+ """
+ resp = self.client.put(
+ f"{self.get_suffix(Suggestion)}/{str(suggestion.id.__root__)}",
+ data=suggestion.json(),
+ )
+
+ return Suggestion(**resp)

ingestion/src/metadata/ingestion/ometa/ometa_api.py

@@ -50,6 +50,7 @@
 from metadata.ingestion.ometa.mixins.search_index_mixin import OMetaSearchIndexMixin
 from metadata.ingestion.ometa.mixins.server_mixin import OMetaServerMixin
 from metadata.ingestion.ometa.mixins.service_mixin import OMetaServiceMixin
+from metadata.ingestion.ometa.mixins.suggestions_mixin import OMetaSuggestionsMixin
 from metadata.ingestion.ometa.mixins.table_mixin import OMetaTableMixin
 from metadata.ingestion.ometa.mixins.tests_mixin import OMetaTestsMixin
 from metadata.ingestion.ometa.mixins.topic_mixin import OMetaTopicMixin
@@ -108,6 +109,7 @@ class OpenMetadata(
  OMetaRolePolicyMixin,
  OMetaSearchIndexMixin,
  OMetaCustomPropertyMixin,
+ OMetaSuggestionsMixin,
  Generic[T, C],
 ):
  """
@@ -246,11 +248,9 @@ def get_entity_from_create(self, create: Type[C]) -> Type[T]:
  )
  return entity_class
 
- def create_or_update(self, data: C) -> T:
+ def _create(self, data: C, method: str) -> T:
  """
- We allow CreateEntity for PUT, so we expect a type C.
-
- We PUT to the endpoint and return the Entity generated result
+ Internal logic to run POST vs. PUT
  """
  entity = data.__class__
  is_create = "create" in data.__class__.__name__.lower()
@@ -262,15 +262,23 @@ def create_or_update(self, data: C) -> T:
  raise InvalidEntityException(
  f"PUT operations need a CreateEntity, not {entity}"
  )
- resp = self.client.put(
-  self.get_suffix(entity), data=data.json(encoder=show_secrets_encoder)
- )
+
+ fn = getattr(self.client, method)
+ resp = fn(self.get_suffix(entity), data=data.json(encoder=show_secrets_encoder))
  if not resp:
  raise EmptyPayloadException(
  f"Got an empty response when trying to PUT to {self.get_suffix(entity)}, {data.json()}"
  )
  return entity_class(**resp)
 
+ def create_or_update(self, data: C) -> T:
+ """Run a PUT requesting via create request C"""
+ return self._create(data=data, method="put")
+
+ def create(self, data: C) -> T:
+ """Run a POST requesting via create request C"""
+ return self._create(data=data, method="post")
+
  def get_by_name(
  self,
  entity: Type[T],

ingestion/src/metadata/ingestion/ometa/routes.py

@@ -50,6 +50,7 @@
  CreateDataProductRequest,
 )
 from metadata.generated.schema.api.domains.createDomain import CreateDomainRequest
+from metadata.generated.schema.api.feed.createSuggestion import CreateSuggestionRequest
 from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest
 from metadata.generated.schema.api.policies.createPolicy import CreatePolicyRequest
 from metadata.generated.schema.api.services.createDashboardService import (
@@ -90,6 +91,7 @@
 from metadata.generated.schema.dataInsight.dataInsightChart import DataInsightChart
 from metadata.generated.schema.dataInsight.kpi.kpi import Kpi
 from metadata.generated.schema.entity.automations.workflow import Workflow
+from metadata.generated.schema.entity.bot import Bot
 from metadata.generated.schema.entity.classification.classification import (
  Classification,
 )
@@ -113,6 +115,7 @@
 from metadata.generated.schema.entity.data.topic import Topic
 from metadata.generated.schema.entity.domains.dataProduct import DataProduct
 from metadata.generated.schema.entity.domains.domain import Domain
+from metadata.generated.schema.entity.feed.suggestion import Suggestion
 from metadata.generated.schema.entity.policies.policy import Policy
 from metadata.generated.schema.entity.services.connections.testConnectionDefinition import (
  TestConnectionDefinition,
@@ -181,6 +184,7 @@
  User.__name__: "/users",
  CreateUserRequest.__name__: "/users",
  AuthenticationMechanism.__name__: "/users/auth-mechanism",
+ Bot.__name__: "/bots", # We won't allow bot creation from the client
  # Roles
  Role.__name__: "/roles",
  CreateRoleRequest.__name__: "/roles",
@@ -225,4 +229,7 @@
  CreateDomainRequest.__name__: "/domains",
  DataProduct.__name__: "/dataProducts",
  CreateDataProductRequest.__name__: "/dataProducts",
+ # Suggestions
+ Suggestion.__name__: "/suggestions",
+ CreateSuggestionRequest.__name__: "/suggestions",
 }

ingestion/src/metadata/ingestion/source/database/dbt/dbt_utils.py

@@ -14,6 +14,7 @@
 import traceback
 from typing import Optional, Union
 
+from metadata.generated.schema.entity.data.table import Table
 from metadata.generated.schema.tests.testSuite import TestSuite
 from metadata.generated.schema.type.entityReference import EntityReference
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
@@ -81,7 +82,8 @@ def generate_entity_link(dbt_test):
  manifest_node = dbt_test.get(DbtCommonEnum.MANIFEST_NODE.value)
  entity_link_list = [
  entity_link.get_entity_link(
- table_fqn=table_fqn,
+ Table,
+ fqn=table_fqn,
  column_name=manifest_node.column_name
  if hasattr(manifest_node, "column_name")
  else None,

ingestion/src/metadata/utils/entity_link.py

@@ -13,17 +13,23 @@
 Filter information has been taken from the
 ES indexes definitions
 """
-from typing import List, Optional
+from typing import Any, List, Optional, TypeVar
 
 from antlr4.CommonTokenStream import CommonTokenStream
 from antlr4.error.ErrorStrategy import BailErrorStrategy
 from antlr4.InputStream import InputStream
 from antlr4.tree.Tree import ParseTreeWalker
+from pydantic import BaseModel
 from requests.compat import unquote_plus
 
 from metadata.antlr.split_listener import EntityLinkSplitListener
 from metadata.generated.antlr.EntityLinkLexer import EntityLinkLexer
 from metadata.generated.antlr.EntityLinkParser import EntityLinkParser
+from metadata.generated.schema.entity.data.table import Table
+from metadata.utils.constants import ENTITY_REFERENCE_TYPE_MAP
+from metadata.utils.dispatch import class_register
+
+T = TypeVar("T", bound=BaseModel)
 
 
 class EntityLinkBuildingException(Exception):
@@ -86,16 +92,30 @@ def get_table_or_column_fqn(entity_link: str) -> str:
  )
 
 
-def get_entity_link(table_fqn: str, column_name: Optional[str]) -> str:
+get_entity_link_registry = class_register()
+
+
+def get_entity_link(entity_type: Any, fqn: str, **kwargs) -> str:
  """From table fqn and column name get the entity_link
 
  Args:
- table_fqn: table fqn
- column_name: Optional param to generate entity link with column name
+ entity_type: Entity being built
+ fqn: Entity fqn
  """
 
+ func = get_entity_link_registry.registry.get(entity_type.__name__)
+ if not func:
+ return f"<#E::{ENTITY_REFERENCE_TYPE_MAP[entity_type.__name__]}::{fqn}>"
+
+ return func(fqn, **kwargs)
+
+
+@get_entity_link_registry.add(Table)
+def _(fqn: str, column_name: Optional[str] = None) -> str:
+ """From table fqn and column name get the entity_link"""
+
  if column_name:
- entity_link = f"<#E::table::" f"{table_fqn}" f"::columns::" f"{column_name}>"
+ entity_link = f"<#E::{ENTITY_REFERENCE_TYPE_MAP[Table.__name__]}::{fqn}::columns::{column_name}>"
  else:
- entity_link = f"<#E::table::" f"{table_fqn}>"
+ entity_link = f"<#E::{ENTITY_REFERENCE_TYPE_MAP[Table.__name__]}::{fqn}>"
  return entity_link

ingestion/src/metadata/utils/helpers.py

@@ -31,7 +31,9 @@
 
 from metadata.generated.schema.entity.data.chart import ChartType
 from metadata.generated.schema.entity.data.table import Column, Table
+from metadata.generated.schema.entity.feed.suggestion import Suggestion, SuggestionType
 from metadata.generated.schema.entity.services.databaseService import DatabaseService
+from metadata.generated.schema.type.basic import EntityLink
 from metadata.generated.schema.type.tagLabel import TagLabel
 from metadata.utils.constants import DEFAULT_DATABASE
 from metadata.utils.logger import utils_logger
@@ -227,12 +229,38 @@ def find_in_iter(element: Any, container: Iterable[Any]) -> Optional[Any]:
  return next((elem for elem in container if elem == element), None)
 
 
-def find_column_in_table(column_name: str, table: Table) -> Optional[Column]:
+def find_column_in_table(
+ column_name: str, table: Table, case_sensitive: bool = True
+) -> Optional[Column]:
  """
  If the column exists in the table, return it
  """
+
+ def equals(first: str, second: str) -> bool:
+ if case_sensitive:
+ return first == second
+ return first.lower() == second.lower()
+
+ return next(
+ (col for col in table.columns if equals(col.name.__root__, column_name)), None
+ )
+
+
+def find_suggestion(
+ suggestions: List[Suggestion],
+ suggestion_type: SuggestionType,
+ entity_link: EntityLink,
+) -> Optional[Suggestion]:
+ """Given a list of suggestions, a suggestion type and an entity link, find
+ one suggestion in the list that matches the criteria
+ """
  return next(
- (col for col in table.columns if col.name.__root__ == column_name), None
+ (
+ sugg
+ for sugg in suggestions
+ if sugg.type == suggestion_type and sugg.entityLink == entity_link
+ ),
+ None,
  )