Add from and size parameters to terms lookup query for pagination support

CHANGELOG.md

@@ -23,6 +23,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add intra segment support for single-value metric aggregations ([#20503](https://github.com/opensearch-project/OpenSearch/pull/20503))
 - Add ref_path support for package-based hunspell dictionary loading ([#20840](https://github.com/opensearch-project/OpenSearch/pull/20840))
 - Add support for enabling pluggable data formats, starting with phase-1 of decoupling shard from engine, and introducing basic abstractions ([#20675](https://github.com/opensearch-project/OpenSearch/pull/20675))
+- Add `from` and `size` parameters to terms lookup query to support pagination of lookup terms ([#20923](https://github.com/opensearch-project/OpenSearch/pull/20923))
 
 ### Changed
 - Make telemetry `Tags` immutable ([#20788](https://github.com/opensearch-project/OpenSearch/pull/20788))

rest-api-spec/src/main/resources/rest-api-spec/test/search/171_terms_lookup_query.yml

@@ -210,3 +210,74 @@
           query:
             match_all: {}
   - match: { hits.total: 5 } # Should always be 5
+
+  # --- TEST CASES FOR from AND size PARAMETERS --- #
+
+  # Test: size parameter limits the number of terms fetched
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: main_index
+        body:
+          query:
+            terms:
+              user:
+                index: lookup_index
+                path: followers
+                query:
+                  term:
+                    group: "g1"
+                size: 1
+  - match: { hits.total: 3 } # size=1 returns 3 docs: u1(foo), u2(bar), u5(foo)
+
+  # Test: from parameter skips the first N terms
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: main_index
+        body:
+          query:
+            terms:
+              user:
+                index: lookup_index
+                path: followers
+                query:
+                  term:
+                    group: "g1"
+                from: 1
+  - match: { hits.total: 1 } # from=1 returns 1 doc: u3(baz)
+
+  # Test: from and size combined
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: main_index
+        body:
+          query:
+            terms:
+              user:
+                index: lookup_index
+                path: followers
+                query:
+                  term:
+                    group: "g1"
+                from: 1
+                size: 1
+  - match: { hits.total: 1 } # from=1, size=1 returns 1 doc: u3(baz)
+
+  # Test: size=0 should return no terms
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: main_index
+        body:
+          query:
+            terms:
+              user:
+                index: lookup_index
+                path: followers
+                query:
+                  term:
+                    group: "g1"
+                size: 0
+  - match: { hits.total: 0 }

server/src/main/java/org/opensearch/index/query/TermsQueryBuilder.java

@@ -616,7 +616,17 @@ private void fetch(TermsLookup termsLookup, Client client, ActionListener<List<O
                         int fetchSize = Math.min(Math.min(maxTermsCount, maxResultWindow), maxClauseCount);
 
                         try {
-                            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().query(termsLookup.query()).size(fetchSize);
+                            // Use user-specified size if provided, otherwise use calculated fetchSize
+                            int effectiveSize = termsLookup.size() != null ? termsLookup.size() : fetchSize;
+                            // Validate that effectiveSize doesn't exceed fetchSize
+                            if (effectiveSize > fetchSize) {
+                                throw new IllegalArgumentException(
+                                    "Terms lookup size [" + effectiveSize + "] exceeds maximum allowed [" + fetchSize + "]"
+                                );
+                            }
+                            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().query(termsLookup.query())
+                                .from(termsLookup.from())
+                                .size(effectiveSize);
 
                             // Use stored fields if possible, otherwise fetch source
                             if (termsLookup.store()) {

server/src/main/java/org/opensearch/indices/TermsLookup.java

@@ -64,12 +64,18 @@ public class TermsLookup implements Writeable, ToXContentFragment {
     private final String path;
     private String routing;
     private QueryBuilder query;
+    private int from = 0;
+    private Integer size;
 
     public TermsLookup(String index, String id, String path) {
         this(index, id, path, null);
     }
 
     public TermsLookup(String index, String id, String path, QueryBuilder query) {
+        this(index, id, path, query, 0, null);
+    }
+
+    public TermsLookup(String index, String id, String path, QueryBuilder query, int from, Integer size) {
         if (index == null) {
             throw new IllegalArgumentException("[" + TermsQueryBuilder.NAME + "] index cannot be null or empty for TermsLookup");
         }
@@ -90,6 +96,8 @@ public TermsLookup(String index, String id, String path, QueryBuilder query) {
         this.id = id;
         this.path = path;
         this.query = query;
+        this.from = from;
+        this.size = size;
     }
 
     public String index() {
@@ -117,6 +125,10 @@ public TermsLookup(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(Version.V_3_2_0)) {
             query = in.readOptionalWriteable(inStream -> inStream.readNamedWriteable(QueryBuilder.class));
         }
+        if (in.getVersion().onOrAfter(Version.V_3_5_0)) {
+            from = in.readInt();
+            size = in.readOptionalInt();
+        }
     }
 
     @Override
@@ -134,6 +146,10 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_3_2_0)) {
             out.writeOptionalWriteable(query);
         }
+        if (out.getVersion().onOrAfter(Version.V_3_5_0)) {
+            out.writeInt(from);
+            out.writeOptionalInt(size);
+        }
     }
 
     public String path() {
@@ -176,6 +192,30 @@ public void setQuery(QueryBuilder query) {
         this.query = query;
     }
 
+    public int from() {
+        return from;
+    }
+
+    public TermsLookup from(int from) {
+        if (from < 0) {
+            throw new IllegalArgumentException("[" + TermsQueryBuilder.NAME + "] from cannot be negative.");
+        }
+        this.from = from;
+        return this;
+    }
+
+    public Integer size() {
+        return size;
+    }
+
+    public TermsLookup size(Integer size) {
+        if (size != null && size < 0) {
+            throw new IllegalArgumentException("[" + TermsQueryBuilder.NAME + "] size cannot be negative.");
+        }
+        this.size = size;
+        return this;
+    }
+
     public TermsLookup id(String id) {
         if (this.query != null && id != null) {
             throw new IllegalArgumentException("[" + TermsQueryBuilder.NAME + "] query lookup element cannot specify both id and query.");
@@ -189,6 +229,8 @@ public TermsLookup id(String id) {
         String id = (String) args[1]; // Optional id or query but not both
         String path = (String) args[2];
         QueryBuilder query = (QueryBuilder) args[3]; // Optional id or query but not both
+        Integer from = (Integer) args[4]; // Optional
+        Integer size = (Integer) args[5]; // Optional
 
         // Validation: Either id or query must be present, but not both
         if (id == null && query == null) {
@@ -199,7 +241,7 @@ public TermsLookup id(String id) {
         if (id != null && query != null) {
             throw new IllegalArgumentException("[" + TermsQueryBuilder.NAME + "] query lookup element cannot specify both id and query.");
         }
-        return new TermsLookup(index, id, path, query);
+        return new TermsLookup(index, id, path, query, from != null ? from : 0, size);
     });
     static {
         PARSER.declareString(constructorArg(), new ParseField("index")); // Required
@@ -212,6 +254,8 @@ public TermsLookup id(String id) {
                 throw new RuntimeException("Error parsing inner query builder", e);
             }
         }, new ParseField("query")); // Optional
+        PARSER.declareInt(optionalConstructorArg(), new ParseField("from")); // Optional
+        PARSER.declareInt(optionalConstructorArg(), new ParseField("size")); // Optional
         PARSER.declareString(TermsLookup::routing, new ParseField("routing")); // Optional
         PARSER.declareBoolean(TermsLookup::store, new ParseField("store")); // Optional
     }
@@ -239,6 +283,12 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field("query");
             query.toXContent(builder, params);
         }
+        if (from > 0) {
+            builder.field("from", from);
+        }
+        if (size != null) {
+            builder.field("size", size);
+        }
         if (store) {
             builder.field("store", true);
         }
@@ -247,7 +297,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
 
     @Override
     public int hashCode() {
-        return Objects.hash(index, id, path, routing, store, query);
+        return Objects.hash(index, id, path, routing, store, query, from, size);
     }
 
     @Override
@@ -264,6 +314,8 @@ public boolean equals(Object obj) {
             && Objects.equals(path, other.path)
             && Objects.equals(routing, other.routing)
             && Objects.equals(store, other.store)
-            && Objects.equals(query, other.query);
+            && Objects.equals(query, other.query)
+            && from == other.from
+            && Objects.equals(size, other.size);
     }
 }