feat(search): include timestamp for entity metadata change

Author: deepgarg-visa

entity-registry/src/main/java/com/linkedin/metadata/models/SearchableFieldSpecExtractor.java

@@ -176,7 +176,9 @@ private void extractSearchableAnnotation(
                 annotation.getNumValuesFieldName(),
                 annotation.getWeightsPerFieldValue(),
                 annotation.getFieldNameAliases(),
-                annotation.isIncludeQueryEmptyAggregation());
+                annotation.isIncludeQueryEmptyAggregation(),
+                annotation.isIncludeSystemModifiedAt(),
+                annotation.getSystemModifiedAtFieldName());
       }
     }
     log.debug("Searchable annotation for field: {} : {}", schemaPathSpec, annotation);

entity-registry/src/main/java/com/linkedin/metadata/models/annotation/SearchableAnnotation.java

@@ -60,6 +60,10 @@ public class SearchableAnnotation {
   // only adds to query time not mapping
   boolean includeQueryEmptyAggregation;
 
+  boolean includeSystemModifiedAt;
+
+  Optional<String> systemModifiedAtFieldName;
+
   public enum FieldType {
     KEYWORD,
     TEXT,
@@ -125,6 +129,10 @@ public static SearchableAnnotation fromPegasusAnnotationObject(
     final List<String> fieldNameAliases = getFieldNameAliases(map);
 
     final FieldType resolvedFieldType = getFieldType(fieldType, schemaDataType);
+    final Optional<Boolean> includeSystemModifiedAt =
+        AnnotationUtils.getField(map, "includeSystemModifiedAt", Boolean.class);
+    final Optional<String> systemModifiedAtFieldName =
+        AnnotationUtils.getField(map, "systemModifiedAtFieldName", String.class);
     return new SearchableAnnotation(
         fieldName.orElse(schemaFieldName),
         resolvedFieldType,
@@ -139,7 +147,9 @@ public static SearchableAnnotation fromPegasusAnnotationObject(
         numValuesFieldName,
         weightsPerFieldValueMap.orElse(ImmutableMap.of()),
         fieldNameAliases,
-        includeQueryEmptyAggregation.orElse(false));
+        includeQueryEmptyAggregation.orElse(false),
+        includeSystemModifiedAt.orElse(false),
+        systemModifiedAtFieldName);
   }
 
   private static FieldType getFieldType(

metadata-io/src/main/java/com/linkedin/metadata/search/elasticsearch/indexbuilder/MappingsBuilder.java

@@ -277,6 +277,17 @@ private static Map<String, Object> getMappingsForField(
         .getNumValuesFieldName()
         .ifPresent(
             fieldName -> mappings.put(fieldName, ImmutableMap.of(TYPE, ESUtils.LONG_FIELD_TYPE)));
+
+    final String fieldName = searchableFieldSpec.getSearchableAnnotation().getFieldName();
+    if (searchableFieldSpec.getSearchableAnnotation().isIncludeSystemModifiedAt()) {
+      String modifiedAtFieldName =
+          searchableFieldSpec
+              .getSearchableAnnotation()
+              .getSystemModifiedAtFieldName()
+              .orElse(String.format("%sSystemModifiedAt", fieldName));
+      mappings.put(modifiedAtFieldName, ImmutableMap.of(TYPE, ESUtils.DATE_FIELD_TYPE));
+    }
+
     mappings.putAll(getMappingsForFieldNameAliases(searchableFieldSpec));
 
     return mappings;

metadata-io/src/main/java/com/linkedin/metadata/search/transformer/SearchDocumentTransformer.java

@@ -255,6 +255,17 @@ public void setSearchableValue(
       return;
     }
 
+    if (fieldSpec.getSearchableAnnotation().isIncludeSystemModifiedAt()) {
+      String modifiedAtFieldName =
+          fieldSpec
+              .getSearchableAnnotation()
+              .getSystemModifiedAtFieldName()
+              .orElse(String.format("%sSystemModifiedAt", fieldName));
+      searchDocument.set(
+          modifiedAtFieldName,
+          JsonNodeFactory.instance.numberNode((Long) System.currentTimeMillis()));
+    }
+
     if (isArray || (valueType == DataSchema.Type.MAP && !OBJECT_FIELD_TYPES.contains(fieldType))) {
       if (fieldType == FieldType.BROWSE_PATH_V2) {
         String browsePathV2Value = getBrowsePathV2Value(fieldValues);

metadata-models/src/main/pegasus/com/linkedin/businessattribute/BusinessAttributeAssociation.pdl

@@ -5,5 +5,10 @@ record BusinessAttributeAssociation {
   /**
   * Urn of the applied businessAttribute
   */
+  @Searchable = {
+     "fieldName": "schemaFieldBusinessAttribute",
+     "includeSystemModifiedAt": true,
+     "systemModifiedAtFieldName": "schemaFieldBusinessAttributeModifiedAt"
+  }
    businessAttributeUrn: BusinessAttributeUrn
 }

metadata-models/src/main/pegasus/com/linkedin/common/GlossaryTermAssociation.pdl

@@ -16,7 +16,9 @@ record GlossaryTermAssociation {
     "fieldType": "URN",
     "hasValuesFieldName": "hasGlossaryTerms",
     "addToFilters": true,
-    "filterNameOverride": "Glossary Term"
+    "filterNameOverride": "Glossary Term",
+    "includeSystemModifiedAt": true,
+    "systemModifiedAtFieldName": "termsModifiedAt"
   }
   urn: GlossaryTermUrn
 

metadata-models/src/main/pegasus/com/linkedin/schema/EditableSchemaFieldInfo.pdl

@@ -67,7 +67,9 @@ record EditableSchemaFieldInfo {
     "/terms/*/urn": {
       "fieldName": "editedFieldGlossaryTerms",
       "fieldType": "URN",
-      "boostScore": 0.5
+      "boostScore": 0.5,
+      "includeSystemModifiedAt": true,
+      "systemModifiedAtFieldName": "schemaFieldTermsModifiedAt"
     },
     "/terms/*/attribution/time": {
         "fieldName": "editedFieldTermAttributionDates",