fix(lineage) Fix lineage source/dest filtering with explored per hop …

…limit (#10879)
datahub-project · Jul 16, 2024 · f6b7bbc · f6b7bbc
1 parent beae5a4
commit f6b7bbc
Showing 1 changed file with 23 additions and 7 deletions.
diff --git a/metadata-io/src/main/java/com/linkedin/metadata/graph/elastic/ESGraphQueryDAO.java b/metadata-io/src/main/java/com/linkedin/metadata/graph/elastic/ESGraphQueryDAO.java
@@ -177,13 +177,18 @@ private SearchResponse executeGroupByLineageSearchQuery(
     sourceFilterQuery.minimumShouldMatch(1);
     validEdges.stream()
         .filter(pair -> RelationshipDirection.OUTGOING.equals(pair.getValue().getDirection()))
-        .forEach(pair -> sourceFilterQuery.should(getAggregationFilter(pair)));
+        .forEach(
+            pair ->
+                sourceFilterQuery.should(
+                    getAggregationFilter(pair, RelationshipDirection.OUTGOING)));
 
     BoolQueryBuilder destFilterQuery = QueryBuilders.boolQuery();
     destFilterQuery.minimumShouldMatch(1);
     validEdges.stream()
         .filter(pair -> RelationshipDirection.INCOMING.equals(pair.getValue().getDirection()))
-        .forEach(pair -> destFilterQuery.should(getAggregationFilter(pair)));
+        .forEach(
+            pair ->
+                destFilterQuery.should(getAggregationFilter(pair, RelationshipDirection.INCOMING)));
 
     FilterAggregationBuilder sourceRelationshipTypeFilters =
         AggregationBuilders.filter(FILTER_BY_SOURCE_RELATIONSHIP, sourceFilterQuery);
@@ -226,17 +231,28 @@ private SearchResponse executeGroupByLineageSearchQuery(
     }
   }
 
-  private BoolQueryBuilder getAggregationFilter(Pair<String, EdgeInfo> pair) {
+  private BoolQueryBuilder getAggregationFilter(
+      Pair<String, EdgeInfo> pair, RelationshipDirection direction) {
     BoolQueryBuilder subFilter = QueryBuilders.boolQuery();
     TermQueryBuilder relationshipTypeTerm =
-        QueryBuilders.termQuery(RELATIONSHIP_TYPE, pair.getValue().getType());
+        QueryBuilders.termQuery(RELATIONSHIP_TYPE, pair.getValue().getType()).caseInsensitive(true);
     subFilter.must(relationshipTypeTerm);
+
+    String sourceType;
+    String destinationType;
+    if (direction.equals(RelationshipDirection.OUTGOING)) {
+      sourceType = pair.getKey();
+      destinationType = pair.getValue().getOpposingEntityType();
+    } else {
+      sourceType = pair.getValue().getOpposingEntityType();
+      destinationType = pair.getKey();
+    }
+
     TermQueryBuilder sourceTypeTerm =
-        QueryBuilders.termQuery(SOURCE + ".entityType", pair.getKey());
+        QueryBuilders.termQuery(SOURCE + ".entityType", sourceType).caseInsensitive(true);
     subFilter.must(sourceTypeTerm);
     TermQueryBuilder destinationTypeTerm =
-        QueryBuilders.termQuery(
-            DESTINATION + ".entityType", pair.getValue().getOpposingEntityType());
+        QueryBuilders.termQuery(DESTINATION + ".entityType", destinationType).caseInsensitive(true);
     subFilter.must(destinationTypeTerm);
     return subFilter;
   }