[2.0] feat(search): add data_contract_search_index (#34304)

ingestion/src/metadata/utils/constants.py

@@ -21,6 +21,7 @@
 from metadata.generated.schema.entity.data.dashboardDataModel import DashboardDataModel
 from metadata.generated.schema.entity.data.database import Database
 from metadata.generated.schema.entity.data.databaseSchema import DatabaseSchema
+from metadata.generated.schema.entity.data.dataContract import DataContract
 from metadata.generated.schema.entity.data.directory import Directory
 from metadata.generated.schema.entity.data.file import File
 from metadata.generated.schema.entity.data.glossary import Glossary
@@ -168,6 +169,7 @@
     "metric": Metric,
     "glossary": Glossary,
     "glossaryTerm": GlossaryTerm,
+    "dataContract": DataContract,
     "eventSubscription": EventSubscription,
 }
 

ingestion/src/metadata/utils/elasticsearch.py

@@ -29,6 +29,7 @@
 from metadata.generated.schema.entity.data.dashboardDataModel import DashboardDataModel
 from metadata.generated.schema.entity.data.database import Database
 from metadata.generated.schema.entity.data.databaseSchema import DatabaseSchema
+from metadata.generated.schema.entity.data.dataContract import DataContract
 from metadata.generated.schema.entity.data.directory import Directory
 from metadata.generated.schema.entity.data.file import File
 from metadata.generated.schema.entity.data.glossary import Glossary
@@ -80,6 +81,7 @@
     Spreadsheet.__name__: "spreadsheet_search_index",
     Worksheet.__name__: "worksheet_search_index",
     Query.__name__: "query_search_index",
+    DataContract.__name__: "data_contract_search_index",
     ReportData.__name__: "entity_report_data_index",
     Metric.__name__: "metric_search_index",
     "web_analytic_user_activity_report": "web_analytic_user_activity_report_data_index",

ingestion/tests/unit/test_ometa_utils.py

@@ -25,6 +25,7 @@
 from metadata.generated.schema.entity.data.dashboardDataModel import DashboardDataModel
 from metadata.generated.schema.entity.data.database import Database
 from metadata.generated.schema.entity.data.databaseSchema import DatabaseSchema
+from metadata.generated.schema.entity.data.dataContract import DataContract
 from metadata.generated.schema.entity.data.directory import Directory
 from metadata.generated.schema.entity.data.mlmodel import MlModel
 from metadata.generated.schema.entity.data.pipeline import Pipeline
@@ -58,6 +59,7 @@
     model_str,
 )
 from metadata.utils.constants import ENTITY_REFERENCE_CLASS_MAP
+from metadata.utils.elasticsearch import ES_INDEX_MAP
 
 MOCK_TABLE = Table(
     id="c3eb265f-5445-4ad3-ba5e-797d3a3071bb",
@@ -100,6 +102,12 @@ def test_get_entity_type(self):
         self.assertEqual(get_entity_type("hello"), "hello")
         self.assertEqual(get_entity_type(MlModel), "mlmodel")
 
+    def test_get_entity_type_data_contract(self):
+        self.assertEqual(get_entity_type(DataContract), "dataContract")
+
+    def test_es_index_map_data_contract(self):
+        self.assertEqual(ES_INDEX_MAP[DataContract.__name__], "data_contract_search_index")
+
     def test_get_entity_type_camel_case_consistency(self):
         """
         Test that get_entity_type returns camelCase entity types

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/DataContractResourceIT.java

@@ -53,6 +53,7 @@
 import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatus;
 import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatusType;
 import org.openmetadata.schema.entity.teams.Role;
+import org.openmetadata.schema.entity.teams.User;
 import org.openmetadata.schema.tests.TestCase;
 import org.openmetadata.schema.tests.TestCaseParameterValue;
 import org.openmetadata.schema.tests.TestDefinition;
@@ -100,7 +101,7 @@ public class DataContractResourceIT extends BaseEntityIT<DataContract, CreateDat
     supportsDataContract = false;
     supportsDomains = false;
     supportsPatchDomains = false;
-    supportsSearchIndex = false; // DataContract doesn't have a search index
+    supportsSearchIndex = true;
     supportsBulkAPI = false;
     supportsRecognizerFeedback = false;
     supportsConversations = false;
@@ -7916,4 +7917,72 @@ private static OpenMetadataClient userAllowedTo(TestNamespace ns, String kind, R
             new CreateUser().withName(userName).withEmail(email).withRoles(List.of(role.getId())));
     return SdkClients.createClient(email, email, new String[] {});
   }
+
+  @Test
+  void patchOwners_reindexesContract(TestNamespace ns) throws Exception {
+    DataContract contract = createEntity(createMinimalRequest(ns));
+    String ownerName = ("dc_owner_" + ns.uniqueShortId()).toLowerCase(Locale.ROOT);
+    User owner =
+        SdkClients.adminClient()
+            .users()
+            .create(
+                new CreateUser()
+                    .withName(ownerName)
+                    .withEmail(ownerName + "@test.openmetadata.org"));
+    EntityReference ownerRef = owner.getEntityReference();
+
+    contract.setOwners(List.of(ownerRef));
+    SdkClients.adminClient().dataContracts().update(contract.getId().toString(), contract);
+
+    String assetFqn = contract.getEntity().getFullyQualifiedName();
+    Awaitility.await("contract doc reflects new owner and asset")
+        .pollInterval(Duration.ofSeconds(2))
+        .atMost(Duration.ofSeconds(120))
+        .ignoreExceptions()
+        .untilAsserted(
+            () -> {
+              String response = searchForEntity(contract.getId().toString());
+              assertTrue(response.contains(ownerRef.getId().toString()), "owner id indexed");
+              assertTrue(response.contains(assetFqn), "entity.fullyQualifiedName indexed");
+              assertTrue(response.contains("\"type\":\"table\""), "entity.type indexed");
+            });
+  }
+
+  @Test
+  void addResult_keepsOwnersInSearchDoc(TestNamespace ns) throws Exception {
+    DataContract contract = createEntity(createMinimalRequest(ns));
+    String ownerName = ("dc_result_owner_" + ns.uniqueShortId()).toLowerCase(Locale.ROOT);
+    User owner =
+        SdkClients.adminClient()
+            .users()
+            .create(
+                new CreateUser()
+                    .withName(ownerName)
+                    .withEmail(ownerName + "@test.openmetadata.org"));
+    String ownerId = owner.getId().toString();
+    contract.setOwners(List.of(owner.getEntityReference()));
+    SdkClients.adminClient().dataContracts().update(contract.getId().toString(), contract);
+
+    String marker = "keep-owners-" + ns.uniqueShortId();
+    SdkClients.adminClient()
+        .dataContracts()
+        .addResult(
+            contract.getId(),
+            new DataContractResult()
+                .withDataContractFQN(contract.getFullyQualifiedName())
+                .withTimestamp(System.currentTimeMillis())
+                .withContractExecutionStatus(ContractExecutionStatus.Failed)
+                .withResult(marker)
+                .withExecutionTime(1L));
+
+    Awaitility.await("contract doc reflects the new result")
+        .pollInterval(Duration.ofSeconds(2))
+        .atMost(Duration.ofSeconds(120))
+        .ignoreExceptions()
+        .untilAsserted(
+            () -> assertTrue(searchForEntity(contract.getId().toString()).contains(marker)));
+    assertTrue(
+        searchForEntity(contract.getId().toString()).contains(ownerId),
+        "recording a result must not drop owners from the search doc");
+  }
 }

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DataContractRepository.java

@@ -128,6 +128,7 @@ public DataContractRepository(OpenMetadataApplicationConfig config) {
         DATA_CONTRACT_UPDATE_FIELDS);
     this.ingestionPipelineMapper = new IngestionPipelineMapper(config);
     this.openMetadataApplicationConfig = config;
+    supportsSearch = true;
   }
 
   @Override
@@ -1596,15 +1597,23 @@ private boolean semanticsRuleMatch(SemanticsRule rule1, SemanticsRule rule2) {
 
   private void updateLatestResult(DataContract dataContract, DataContractResult result) {
     try {
-      DataContract updated = JsonUtils.deepCopy(dataContract, DataContract.class);
+      // Reload with the relationship fields the search doc is built from: callers pass contracts
+      // fetched without owners/reviewers, and the reindex after this update would blank them.
+      DataContract original =
+          Entity.getEntity(
+              Entity.DATA_CONTRACT,
+              dataContract.getId(),
+              "owners,reviewers,extension",
+              Include.NON_DELETED);
+      DataContract updated = JsonUtils.deepCopy(original, DataContract.class);
       updated.setLatestResult(
           new LatestResult()
               .withTimestamp(result.getTimestamp())
               .withStatus(result.getContractExecutionStatus())
               .withMessage(result.getResult())
               .withResultId(result.getId()));
       EntityRepository.EntityUpdater entityUpdater =
-          getUpdater(dataContract, updated, EntityRepository.Operation.PATCH, null);
+          getUpdater(original, updated, EntityRepository.Operation.PATCH, null);
       entityUpdater.update();
     } catch (Exception e) {
       LOG.error(

openmetadata-service/src/main/java/org/openmetadata/service/search/SearchIndexFactory.java

@@ -21,6 +21,7 @@
 import org.openmetadata.schema.entity.data.ContextFile;
 import org.openmetadata.schema.entity.data.Dashboard;
 import org.openmetadata.schema.entity.data.DashboardDataModel;
+import org.openmetadata.schema.entity.data.DataContract;
 import org.openmetadata.schema.entity.data.Database;
 import org.openmetadata.schema.entity.data.DatabaseSchema;
 import org.openmetadata.schema.entity.data.Directory;
@@ -79,6 +80,7 @@
 import org.openmetadata.service.search.indexes.DashboardDataModelIndex;
 import org.openmetadata.service.search.indexes.DashboardIndex;
 import org.openmetadata.service.search.indexes.DashboardServiceIndex;
+import org.openmetadata.service.search.indexes.DataContractIndex;
 import org.openmetadata.service.search.indexes.DataProductIndex;
 import org.openmetadata.service.search.indexes.DatabaseIndex;
 import org.openmetadata.service.search.indexes.DatabaseSchemaIndex;
@@ -215,6 +217,7 @@ public SearchIndex buildIndex(String entityType, Object entity) {
       case Entity.CONTEXT_FILE -> new ContextFileIndex((ContextFile) entity);
       case Entity.CONTEXT_MEMORY -> new ContextMemoryIndex((ContextMemory) entity);
       case Entity.PAGE -> new PageIndex((Page) entity);
+      case Entity.DATA_CONTRACT -> new DataContractIndex((DataContract) entity);
       case Entity.DATA_PRODUCT -> new DataProductIndex((DataProduct) entity);
       case Entity.METADATA_SERVICE -> new MetadataServiceIndex((MetadataService) entity);
       case Entity.ENTITY_REPORT_DATA -> new EntityReportDataIndex((ReportData) entity);

openmetadata-service/src/main/java/org/openmetadata/service/search/indexes/DataContractIndex.java

@@ -0,0 +1,47 @@
+package org.openmetadata.service.search.indexes;
+
+import java.util.Map;
+import java.util.Set;
+import org.openmetadata.schema.entity.data.DataContract;
+import org.openmetadata.service.Entity;
+
+public record DataContractIndex(DataContract dataContract) implements SearchIndex {
+  /** Large or free-form contract blocks; indexing them would explode the dynamic mapping. */
+  private static final Set<String> EXCLUDED_FIELDS =
+      Set.of(
+          "schema",
+          "semantics",
+          "qualityExpectations",
+          "odcsQualityRules",
+          "odcsElementExtensions",
+          "contractUpdates",
+          "sla",
+          "security",
+          "termsOfUse");
+
+  @Override
+  public Object getEntity() {
+    return dataContract;
+  }
+
+  @Override
+  public String getEntityTypeName() {
+    return Entity.DATA_CONTRACT;
+  }
+
+  @Override
+  public Map<String, Object> buildSearchIndexDocInternal(Map<String, Object> doc) {
+    doc.put("entity", getEntityWithDisplayName(dataContract.getEntity()));
+    doc.put("testSuite", getEntityWithDisplayName(dataContract.getTestSuite()));
+    return doc;
+  }
+
+  @Override
+  public Set<String> getExcludedFields() {
+    return EXCLUDED_FIELDS;
+  }
+
+  public static Map<String, Float> getFields() {
+    return SearchIndex.getDefaultFields();
+  }
+}

openmetadata-service/src/test/java/org/openmetadata/service/search/DataContractIndexMappingTest.java

@@ -0,0 +1,107 @@
+package org.openmetadata.service.search;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class DataContractIndexMappingTest {
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+  private static final List<String> FREE_FORM_FIELDS =
+      List.of(
+          "schema",
+          "semantics",
+          "qualityExpectations",
+          "odcsQualityRules",
+          "odcsElementExtensions",
+          "contractUpdates",
+          "sla",
+          "security",
+          "termsOfUse");
+
+  private static JsonNode read(String path) throws IOException {
+    try (InputStream in = DataContractIndexMappingTest.class.getResourceAsStream(path)) {
+      assertNotNull(in, "missing resource " + path);
+      return MAPPER.readTree(in);
+    }
+  }
+
+  private static JsonNode properties(String language) throws IOException {
+    return read("/elasticsearch/" + language + "/data_contract_index_mapping.json")
+        .path("mappings")
+        .path("properties");
+  }
+
+  @Test
+  void indexIsRegisteredWithoutParentAliases() throws IOException {
+    JsonNode entry = read("/elasticsearch/indexMapping.json").path("dataContract");
+    assertEquals("data_contract_search_index", entry.path("indexName").asText());
+    assertEquals(
+        "/elasticsearch/%s/data_contract_index_mapping.json",
+        entry.path("indexMappingFile").asText());
+    assertEquals("dataContract", entry.path("alias").asText());
+    // Not under `all`: owner/team asset lists and the domain/tag asset pickers query `all`, and a
+    // contract there would be listed with no link and accepted as a domain/tag asset it can't hold.
+    assertTrue(entry.path("parentAliases").isEmpty());
+    assertTrue(entry.path("childAliases").isEmpty());
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"en", "jp", "ru", "zh"})
+  void contractFieldsAreMapped(String language) throws IOException {
+    JsonNode props = properties(language);
+    assertEquals("keyword", props.at("/entity/properties/type/type").asText());
+    assertEquals(
+        "keyword", props.at("/entity/properties/fullyQualifiedName/fields/keyword/type").asText());
+    assertEquals("keyword", props.at("/testSuite/properties/id/type").asText());
+    assertEquals("keyword", props.at("/latestResult/properties/status/type").asText());
+    assertEquals("keyword", props.at("/reviewers/properties/id/type").asText());
+    assertEquals("nested", props.at("/owners/type").asText());
+    assertEquals("keyword", props.at("/fqnHash/type").asText());
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"en", "jp", "ru", "zh"})
+  void entityFqnKeywordHasNoIgnoreAbove(String language) throws IOException {
+    JsonNode keyword =
+        properties(language).at("/entity/properties/fullyQualifiedName/fields/keyword");
+    assertTrue(
+        keyword.path("ignore_above").isMissingNode(),
+        "asset FQNs can exceed 256 chars; ignore_above would drop them from the filter");
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"en", "jp", "ru", "zh"})
+  void referenceFqnKeywordsAreLowercaseNormalized(String language) throws IOException {
+    // Aggregation autocomplete lowercases the typed prefix; an un-normalized keyword never matches
+    // mixed-case FQNs once the user types.
+    JsonNode props = properties(language);
+    for (String ref : List.of("entity", "testSuite")) {
+      assertEquals(
+          "lowercase_normalizer",
+          props
+              .at("/" + ref + "/properties/fullyQualifiedName/fields/keyword/normalizer")
+              .asText());
+    }
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"en", "jp", "ru", "zh"})
+  void freeFormBlocksAndTagsAreNotMapped(String language) throws IOException {
+    JsonNode props = properties(language);
+    for (String field : FREE_FORM_FIELDS) {
+      assertFalse(props.has(field), field + " must be excluded, not mapped");
+    }
+    assertFalse(props.has("tags"), "contracts have no tags");
+    assertFalse(props.has("classificationTags"), "contracts have no tags");
+  }
+}

openmetadata-service/src/test/java/org/openmetadata/service/search/SearchIndexFactoryTest.java

@@ -34,6 +34,7 @@
 import org.openmetadata.schema.entity.data.Container;
 import org.openmetadata.schema.entity.data.Dashboard;
 import org.openmetadata.schema.entity.data.DashboardDataModel;
+import org.openmetadata.schema.entity.data.DataContract;
 import org.openmetadata.schema.entity.data.Database;
 import org.openmetadata.schema.entity.data.DatabaseSchema;
 import org.openmetadata.schema.entity.data.Directory;
@@ -91,6 +92,7 @@
 import org.openmetadata.service.search.indexes.DashboardDataModelIndex;
 import org.openmetadata.service.search.indexes.DashboardIndex;
 import org.openmetadata.service.search.indexes.DashboardServiceIndex;
+import org.openmetadata.service.search.indexes.DataContractIndex;
 import org.openmetadata.service.search.indexes.DataProductIndex;
 import org.openmetadata.service.search.indexes.DatabaseIndex;
 import org.openmetadata.service.search.indexes.DatabaseSchemaIndex;
@@ -416,6 +418,8 @@ private static Stream<Arguments> supportedIndexMappings() {
             AiFrameworkControlIndex.class),
         Arguments.of(
             Entity.AUDIT_REPORT, (Supplier<Object>) AuditReport::new, AuditReportIndex.class),
+        Arguments.of(
+            Entity.DATA_CONTRACT, (Supplier<Object>) DataContract::new, DataContractIndex.class),
         Arguments.of(Entity.TAG, (Supplier<Object>) Tag::new, TagIndex.class),
         Arguments.of(
             Entity.CLASSIFICATION,