[feat] history relationship dedup logic (#538)

* dedup logic

* fix + unit test

* fix

* utilize validator

Author: ybz1013

dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/EbeanLocalRelationshipQueryDAO.java

@@ -522,26 +522,54 @@ private <SNAPSHOT extends RecordTemplate> SNAPSHOT constructSnapshot(@Nonnull fi
    * INNER JOIN source_entity_table st ON st.urn = rt.sourceEntityUrn
    * WHERE destination entity filters AND source entity filters AND relationship filters</p>
    *
+   * <p> or if relationshipLookUpContext.isIncludeNonCurrentRelationships is true </p>
+   *
+   * <p>SELECT * FROM (
+   * SELECT rt.*, ROW_NUMBER() OVER (PARTITION BY rt.source, rt.metadata$type, rt.destination ORDER BY rt.lastmodifiedon DESC) AS row_num
+   * FROM relationship_table rt
+   * INNER JOIN destination_entity_table dt ON dt.urn = rt.destinationEntityUrn
+   * INNER JOIN source_entity_table st ON st.urn = rt.sourceEntityUrn
+   * WHERE destination entity filters AND source entity filters AND relationship filters)
+   * ranked_rows WHERE row_num = 1</p>
+   *
    * @param relationshipTableName   relationship table name
    * @param relationshipFilter      filter on relationship
    * @param sourceTableName         source entity table name
    * @param sourceEntityFilter      filter on source entity.
    * @param destTableName           destination entity table name. Always null if building relationship with non-mg
    *                                entity.
    * @param destinationEntityFilter filter on destination entity.
-   * @param limit                   max number of records to return. If < 0, will return all records.
-   * @param offset                  offset to start from. If < 0, will start from 0.
+   * @param limit                   max number of records to return. If less than 0, will return all records.
+   * @param offset                  offset to start from. If less than 0, will start from 0.
    */
   @Nonnull
-  private String buildFindRelationshipSQL(
-      @Nonnull final String relationshipTableName, @Nonnull final LocalRelationshipFilter relationshipFilter,
-      @Nullable final String sourceTableName, @Nullable final LocalRelationshipFilter sourceEntityFilter,
-      @Nullable final String destTableName, @Nullable final LocalRelationshipFilter destinationEntityFilter,
-      int limit, int offset, RelationshipLookUpContext relationshipLookUpContext) {
+  @VisibleForTesting
+  public String buildFindRelationshipSQL(@Nonnull final String relationshipTableName,
+      @Nonnull final LocalRelationshipFilter relationshipFilter, @Nullable final String sourceTableName,
+      @Nullable final LocalRelationshipFilter sourceEntityFilter, @Nullable final String destTableName,
+      @Nullable final LocalRelationshipFilter destinationEntityFilter, int limit, int offset,
+      RelationshipLookUpContext relationshipLookUpContext) {
 
     boolean includeNonCurrentRelationships = relationshipLookUpContext.isIncludeNonCurrentRelationships();
     StringBuilder sqlBuilder = new StringBuilder();
-    sqlBuilder.append("SELECT rt.* FROM ").append(relationshipTableName).append(" rt ");
+
+    if (includeNonCurrentRelationships) {
+      sqlBuilder.append("SELECT * FROM (");
+    }
+
+    sqlBuilder.append("SELECT rt.*");
+
+    if (includeNonCurrentRelationships) {
+      final boolean isNonDollarVirtualColumnsEnabled = _eBeanDAOConfig.isNonDollarVirtualColumnsEnabled();
+      final String metadataTypeColName = isNonDollarVirtualColumnsEnabled ? "metadata0type" : "metadata$type";
+      final boolean hasMetadataTypeCol = _schemaValidatorUtil.columnExists(relationshipTableName, metadataTypeColName);
+
+      sqlBuilder.append(", ROW_NUMBER() OVER (PARTITION BY rt.source")
+          .append(hasMetadataTypeCol ? ", rt." + metadataTypeColName : "")
+          .append(", rt.destination ORDER BY rt.lastmodifiedon DESC) AS row_num");
+    }
+
+    sqlBuilder.append(" FROM ").append(relationshipTableName).append(" rt ");
 
     List<Pair<LocalRelationshipFilter, String>> filters = new ArrayList<>();
 
@@ -612,7 +640,8 @@ private String buildFindRelationshipSQL(
       }
       if (whereClauseBuilder.length() != 0) {
         if (includeNonCurrentRelationships) {
-          sqlBuilder.append("WHERE ").append(whereClauseBuilder.toString().replaceFirst("^AND\\s+", ""));
+          String where = whereClauseBuilder.toString().replaceFirst("\\s*AND\\s+", "");
+          sqlBuilder.append("WHERE ").append(where);
         } else {
           sqlBuilder.append("WHERE ").append(whereClauseBuilder);
         }
@@ -628,6 +657,11 @@ private String buildFindRelationshipSQL(
         sqlBuilder.append(" OFFSET ").append(offset);
       }
     }
+
+    if (includeNonCurrentRelationships) {
+      sqlBuilder.append(") ranked_rows WHERE row_num = 1");
+    }
+
     return sqlBuilder.toString();
   }