Route pushDownSort count through LIMIT > k validation

pushDownSort() called requestBuilder.pushDownLimit() directly, bypassing
the LIMIT > k guard in pushDownLimit(). Extract validateLimitWithinK()
helper and call it from both paths so the invariant holds when PPL or
future callers set a non-zero sort count.

Signed-off-by: Eric Wei <mengwei.eric@gmail.com>

Author: mengweieric

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/VectorSearchQueryBuilder.java

@@ -89,13 +89,7 @@ public boolean pushDownFilter(LogicalFilter filter) {
 
   @Override
   public boolean pushDownLimit(LogicalLimit limit) {
-    if (options.containsKey("k")) {
-      int k = Integer.parseInt(options.get("k"));
-      if (limit.getLimit() > k) {
-        throw new ExpressionEvaluationException(
-            String.format("LIMIT %d exceeds k=%d in top-k vector search", limit.getLimit(), k));
-      }
-    }
+    validateLimitWithinK(limit.getLimit());
     limitPushed = true;
     return super.pushDownLimit(limit);
   }
@@ -123,11 +117,23 @@ public boolean pushDownSort(LogicalSort sort) {
     // Preserve the parent's sort.getCount() → limit pushdown contract: SQL always sets count=0,
     // but PPL or future callers may set a non-zero count to combine sort+limit in one node.
     if (sort.getCount() != 0) {
+      validateLimitWithinK(sort.getCount());
       requestBuilder.pushDownLimit(sort.getCount(), 0);
     }
     return true;
   }
 
+  /** Validates that the requested limit does not exceed k in top-k mode. */
+  private void validateLimitWithinK(int limit) {
+    if (options.containsKey("k")) {
+      int k = Integer.parseInt(options.get("k"));
+      if (limit > k) {
+        throw new ExpressionEvaluationException(
+            String.format("LIMIT %d exceeds k=%d in top-k vector search", limit, k));
+      }
+    }
+  }
+
   @Override
   public OpenSearchRequestBuilder build() {
     if (filterTypeExplicit && !filterPushed) {

opensearch/src/test/java/org/opensearch/sql/opensearch/storage/scan/VectorSearchQueryBuilderTest.java

@@ -183,6 +183,28 @@ void pushDownSortPreservesSortCountAsLimit() {
         "sort.getCount() should be pushed down as request size");
   }
 
+  @Test
+  void pushDownSortCountExceedingKRejects() {
+    var requestBuilder = createRequestBuilder();
+    var knnQuery = new WrapperQueryBuilder("{\"knn\":{}}");
+    var builder = new VectorSearchQueryBuilder(requestBuilder, knnQuery, Map.of("k", "5"));
+
+    var dummyChild = new LogicalValues(Collections.emptyList());
+    // LogicalSort with count=10 exceeds k=5 — should be rejected
+    var sort =
+        new org.opensearch.sql.planner.logical.LogicalSort(
+            dummyChild,
+            10,
+            List.of(
+                org.apache.commons.lang3.tuple.ImmutablePair.of(
+                    org.opensearch.sql.ast.tree.Sort.SortOption.DEFAULT_DESC,
+                    new ReferenceExpression("_score", ExprCoreType.FLOAT))));
+
+    ExpressionEvaluationException ex =
+        assertThrows(ExpressionEvaluationException.class, () -> builder.pushDownSort(sort));
+    assertTrue(ex.getMessage().contains("LIMIT 10 exceeds k=5"));
+  }
+
   @Test
   void pushDownSortNonScoreFieldRejected() {
     var requestBuilder = createRequestBuilder();