fix(graph-dac): push indexed equality filters into the native search query

ontology-engine/graph-dac-api/src/main/java/org/sunbird/graph/service/operation/SearchAsyncOperations.java

@@ -4,8 +4,10 @@
 import org.apache.commons.lang3.StringUtils;
 import org.janusgraph.core.JanusGraph;
 import org.janusgraph.core.JanusGraphEdge;
+import org.janusgraph.core.JanusGraphQuery;
 import org.janusgraph.core.JanusGraphTransaction;
 import org.janusgraph.core.JanusGraphVertex;
+import org.sunbird.common.Platform;
 import org.sunbird.common.dto.Property;
 import org.sunbird.common.dto.Request;
 import org.sunbird.common.exception.ClientException;
@@ -33,9 +35,11 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -252,6 +256,11 @@ private static List<Node> executeNativeSearch(JanusGraphTransaction tx, String g
             query.has(SystemProperties.IL_FUNC_OBJECT_TYPE.name(), sc.getObjectType());
         }
 
+        // Callers that express objectType/nodeType as metadata filters rather than via
+        // setObjectType()/setNodeType() would otherwise leave the base query as a bare
+        // has("graphId", ...) - a predicate every node satisfies - forcing a full scan.
+        pushIndexedFilters(query, sc.getMetadata());
+
         // Execute Base Query
         Iterable<JanusGraphVertex> vertices;
         if (!ids.isEmpty()) {
@@ -307,6 +316,57 @@ private static List<Node> executeNativeSearch(JanusGraphTransaction tx, String g
         return new ArrayList<>(nodeList.subList(start, end));
     }
 
+    /**
+     * Property keys backed by a JanusGraph composite index. Pushing a predicate on any other
+     * key gains nothing - JanusGraph would still scan - so the list is restricted to the keys
+     * the graph schema actually indexes. Configurable via {@code graph.native_search.indexed_keys};
+     * an empty list restores the previous (scan) behaviour.
+     */
+    private static final Set<String> INDEXED_KEYS = new HashSet<>(
+            Platform.getStringList("graph.native_search.indexed_keys", Arrays.asList(
+                    SystemProperties.IL_FUNC_OBJECT_TYPE.name(), SystemProperties.IL_SYS_NODE_TYPE.name(),
+                    "code", "identifier", "channel", "framework", "mimeType", "contentType",
+                    "visibility", "status", "createdBy", "objectType", "primaryCategory",
+                    "resourceType", "mediaType", "name", "versionKey")));
+
+    /**
+     * Adds indexable equality predicates from the search metadata to the graph query so that
+     * JanusGraph can resolve them through a composite index instead of returning every vertex
+     * for {@link #matchesMetadata} to filter in memory - which lazy-loads each vertex's
+     * properties from the storage backend one round trip at a time.
+     *
+     * <p>Only conditions that must hold for every match are pushed down:
+     * <ul>
+     *   <li>criteria combined with OR are skipped - an OR branch need not be true, so pushing
+     *       it would exclude valid matches;</li>
+     *   <li>only {@link SearchConditions#OP_EQUAL} is pushed - a composite index cannot serve
+     *       {@code !=}, range or list predicates;</li>
+     *   <li>only keys in {@link #INDEXED_KEYS} with a String value are pushed.</li>
+     * </ul>
+     *
+     * <p>This narrows the candidate set only. In-memory filtering still runs over the result,
+     * so the returned nodes are unchanged.
+     */
+    private static void pushIndexedFilters(JanusGraphQuery query, List<MetadataCriterion> metadata) {
+        if (CollectionUtils.isEmpty(metadata) || INDEXED_KEYS.isEmpty())
+            return;
+        for (MetadataCriterion mc : metadata) {
+            if (null == mc || StringUtils.equalsIgnoreCase(SearchConditions.LOGICAL_OR, mc.getOp()))
+                continue;
+            if (CollectionUtils.isEmpty(mc.getFilters()))
+                continue;
+            for (Filter filter : mc.getFilters()) {
+                if (null == filter || !StringUtils.equals(SearchConditions.OP_EQUAL, filter.getOperator()))
+                    continue;
+                if (!INDEXED_KEYS.contains(filter.getProperty()))
+                    continue;
+                Object value = filter.getValue();
+                if (value instanceof String && StringUtils.isNotBlank((String) value))
+                    query.has(filter.getProperty(), value);
+            }
+        }
+    }
+
     private static void extractIdsFromMetadata(List<MetadataCriterion> metadata, List<String> ids) {
         if (metadata == null)
             return;

ontology-engine/graph-dac-api/src/test/java/org/sunbird/graph/service/operation/SearchAsyncOperationsPushDownTest.java

@@ -0,0 +1,161 @@
+package org.sunbird.graph.service.operation;
+
+import org.janusgraph.core.JanusGraphTransaction;
+import org.janusgraph.core.JanusGraphVertex;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.sunbird.graph.dac.model.Filter;
+import org.sunbird.graph.dac.model.MetadataCriterion;
+import org.sunbird.graph.dac.model.Node;
+import org.sunbird.graph.dac.model.SearchConditions;
+import org.sunbird.graph.dac.model.SearchCriteria;
+import org.sunbird.test.BaseTest;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Covers the indexed-filter push-down in {@link SearchAsyncOperations#executeNativeSearch}.
+ *
+ * <p>The push-down only narrows which vertices the in-memory filter runs over, so every case
+ * here asserts on the returned nodes: pushing predicates down must never change the result.
+ */
+public class SearchAsyncOperationsPushDownTest extends BaseTest {
+
+    private static final String GRAPH_ID = "domain";
+
+    @BeforeClass
+    public static void seed() {
+        JanusGraphTransaction tx = graph.newTransaction();
+        try {
+            addNode(tx, "cat_board", "Category", "DATA_NODE", "Live", "board");
+            addNode(tx, "cat_medium", "Category", "DATA_NODE", "Live", "medium");
+            addNode(tx, "cat_retired", "Category", "DATA_NODE", "Retired", "gradeLevel");
+            addNode(tx, "term_english", "Term", "DATA_NODE", "Live", "english");
+            addNode(tx, "content_one", "Content", "DATA_NODE", "Draft", "course-one");
+            addNode(tx, "def_node", "Category", "DEFINITION_NODE", "Live", "definition");
+            tx.commit();
+        } catch (Exception e) {
+            tx.rollback();
+            throw e;
+        }
+    }
+
+    private static void addNode(JanusGraphTransaction tx, String id, String objectType,
+                                String nodeType, String status, String code) {
+        if (tx.query().has("IL_UNIQUE_ID", id).vertices().iterator().hasNext())
+            return;
+        JanusGraphVertex v = tx.addVertex(GRAPH_ID);
+        v.property("IL_UNIQUE_ID", id);
+        v.property("graphId", GRAPH_ID);
+        v.property("IL_FUNC_OBJECT_TYPE", objectType);
+        v.property("IL_SYS_NODE_TYPE", nodeType);
+        v.property("status", status);
+        v.property("code", code);
+        // Not in the indexed allowlist - exercises the in-memory filter path.
+        v.property("description", code + "-desc");
+    }
+
+    private static List<String> identifiers(SearchCriteria sc) throws Exception {
+        Future<List<Node>> future = SearchAsyncOperations.getNodeByUniqueIds(GRAPH_ID, sc);
+        List<Node> nodes = Await.result(future, Duration.apply("30s"));
+        return nodes.stream().map(Node::getIdentifier).sorted().collect(Collectors.toList());
+    }
+
+    private static SearchCriteria criteria(MetadataCriterion mc) {
+        SearchCriteria sc = new SearchCriteria();
+        sc.addMetadata(mc);
+        sc.setCountQuery(false);
+        return sc;
+    }
+
+    /** The master-category query shape that regressed: equality filters carried as metadata. */
+    @Test
+    public void testAndEqualityFiltersReturnOnlyMatchingNodes() throws Exception {
+        MetadataCriterion mc = MetadataCriterion.create(new ArrayList<Filter>(Arrays.asList(
+                new Filter("IL_FUNC_OBJECT_TYPE", SearchConditions.OP_EQUAL, "Category"),
+                new Filter("IL_SYS_NODE_TYPE", SearchConditions.OP_EQUAL, "DATA_NODE"),
+                new Filter("status", SearchConditions.OP_NOT_EQUAL, "Retired"))));
+
+        Assert.assertEquals(Arrays.asList("cat_board", "cat_medium"), identifiers(criteria(mc)));
+    }
+
+    /**
+     * status is pushed down as an equality, so this also proves a pushed predicate does not
+     * over-filter when combined with others.
+     */
+    @Test
+    public void testEqualityOnPushedKeyNarrowsCorrectly() throws Exception {
+        MetadataCriterion mc = MetadataCriterion.create(new ArrayList<Filter>(Arrays.asList(
+                new Filter("IL_FUNC_OBJECT_TYPE", SearchConditions.OP_EQUAL, "Category"),
+                new Filter("status", SearchConditions.OP_EQUAL, "Retired"))));
+
+        Assert.assertEquals(Arrays.asList("cat_retired"), identifiers(criteria(mc)));
+    }
+
+    /** DEFINITION_NODE must be excluded - the nodeType predicate has to be honoured. */
+    @Test
+    public void testNodeTypeFilterExcludesDefinitionNodes() throws Exception {
+        MetadataCriterion mc = MetadataCriterion.create(new ArrayList<Filter>(Arrays.asList(
+                new Filter("IL_FUNC_OBJECT_TYPE", SearchConditions.OP_EQUAL, "Category"),
+                new Filter("IL_SYS_NODE_TYPE", SearchConditions.OP_EQUAL, "DEFINITION_NODE"))));
+
+        Assert.assertEquals(Arrays.asList("def_node"), identifiers(criteria(mc)));
+    }
+
+    /**
+     * OR criteria must NOT be pushed down: an OR branch need not hold for every match, so
+     * pushing it would drop the nodes that matched the other branch.
+     */
+    @Test
+    public void testOrCriterionIsNotPushedDown() throws Exception {
+        MetadataCriterion mc = MetadataCriterion.create(new ArrayList<Filter>(Arrays.asList(
+                new Filter("IL_FUNC_OBJECT_TYPE", SearchConditions.OP_EQUAL, "Term"),
+                new Filter("IL_FUNC_OBJECT_TYPE", SearchConditions.OP_EQUAL, "Content"))));
+        mc.setOp(SearchConditions.LOGICAL_OR);
+
+        Assert.assertEquals(Arrays.asList("content_one", "term_english"), identifiers(criteria(mc)));
+    }
+
+    /** An indexed key filters correctly once pushed onto the graph query. */
+    @Test
+    public void testIndexedKeyFilterIsCorrect() throws Exception {
+        MetadataCriterion mc = MetadataCriterion.create(new ArrayList<Filter>(Arrays.asList(
+                new Filter("code", SearchConditions.OP_EQUAL, "english"))));
+
+        Assert.assertEquals(Arrays.asList("term_english"), identifiers(criteria(mc)));
+    }
+
+    /** A key outside the indexed allowlist is never pushed, and still filters in memory. */
+    @Test
+    public void testNonIndexedKeyStillFilters() throws Exception {
+        MetadataCriterion mc = MetadataCriterion.create(new ArrayList<Filter>(Arrays.asList(
+                new Filter("description", SearchConditions.OP_EQUAL, "english-desc"))));
+
+        Assert.assertEquals(Arrays.asList("term_english"), identifiers(criteria(mc)));
+    }
+
+    /** Mixed indexed and non-indexed predicates must both be applied. */
+    @Test
+    public void testIndexedAndNonIndexedFiltersCombine() throws Exception {
+        MetadataCriterion mc = MetadataCriterion.create(new ArrayList<Filter>(Arrays.asList(
+                new Filter("IL_FUNC_OBJECT_TYPE", SearchConditions.OP_EQUAL, "Category"),
+                new Filter("description", SearchConditions.OP_EQUAL, "board-desc"))));
+
+        Assert.assertEquals(Arrays.asList("cat_board"), identifiers(criteria(mc)));
+    }
+
+    /** No metadata at all - every node in the graph for this graphId. */
+    @Test
+    public void testNoMetadataReturnsAllNodes() throws Exception {
+        SearchCriteria sc = new SearchCriteria();
+        sc.setCountQuery(false);
+        Assert.assertTrue(identifiers(sc).containsAll(Arrays.asList("cat_board", "term_english", "content_one")));
+    }
+}