Harden input validation and add size=k default for top-k mode

- parseOptions: reject malformed segments and duplicate keys
- parseVector: wrap errors in ExpressionEvaluationException, reject
  non-finite floats (Infinity, NaN)
- VectorSearchIndex: default requestedTotalSize to k via
  pushDownLimitToRequestTotal so queries without LIMIT return k results
- Add 5 new tests: malformed option, duplicate key, empty vector,
  malformed vector component, non-finite vector component

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

Author: mengweieric

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/VectorSearchIndex.java

@@ -53,6 +53,13 @@ public TableScanBuilder createScanBuilder() {
     var queryBuilder = new VectorSearchQueryBuilder(requestBuilder, buildKnnQuery());
     requestBuilder.pushDownTrackedScore(true);
 
+    // Top-k mode: default size to k so queries without LIMIT return k results
+    // instead of falling into the generic large-scan path.
+    // LIMIT pushdown will further reduce this if present.
+    if (options.containsKey("k")) {
+      requestBuilder.pushDownLimitToRequestTotal(Integer.parseInt(options.get("k")), 0);
+    }
+
     Function<OpenSearchRequestBuilder, OpenSearchIndexScan> createScanOperator =
         rb ->
             new OpenSearchIndexScan(

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/VectorSearchTableFunctionImplementation.java

@@ -104,21 +104,44 @@ public Table applyArguments() {
 
   private float[] parseVector(String vectorLiteral) {
     String cleaned = vectorLiteral.replaceAll("[\\[\\]]", "").trim();
+    if (cleaned.isEmpty()) {
+      throw new ExpressionEvaluationException("Vector literal must not be empty");
+    }
     String[] parts = cleaned.split(",");
     float[] vector = new float[parts.length];
     for (int i = 0; i < parts.length; i++) {
-      vector[i] = Float.parseFloat(parts[i].trim());
+      try {
+        vector[i] = Float.parseFloat(parts[i].trim());
+      } catch (NumberFormatException e) {
+        throw new ExpressionEvaluationException(
+            String.format("Invalid vector component '%s': must be a number", parts[i].trim()));
+      }
+      if (!Float.isFinite(vector[i])) {
+        throw new ExpressionEvaluationException(
+            String.format(
+                "Invalid vector component '%s': must be a finite number", parts[i].trim()));
+      }
     }
     return vector;
   }
 
   static Map<String, String> parseOptions(String optionStr) {
     Map<String, String> options = new LinkedHashMap<>();
     for (String pair : optionStr.split(",")) {
-      String[] kv = pair.trim().split("=", 2);
-      if (kv.length == 2) {
-        options.put(kv[0].trim(), kv[1].trim());
+      String trimmed = pair.trim();
+      if (trimmed.isEmpty()) {
+        continue;
+      }
+      String[] kv = trimmed.split("=", 2);
+      if (kv.length != 2 || kv[0].trim().isEmpty() || kv[1].trim().isEmpty()) {
+        throw new ExpressionEvaluationException(
+            String.format("Malformed option segment '%s': expected key=value", trimmed));
+      }
+      String key = kv[0].trim();
+      if (options.containsKey(key)) {
+        throw new ExpressionEvaluationException(String.format("Duplicate option key '%s'", key));
       }
+      options.put(key, kv[1].trim());
     }
     return options;
   }

opensearch/src/test/java/org/opensearch/sql/opensearch/storage/VectorSearchTableFunctionImplementationTest.java

@@ -120,6 +120,51 @@ void testParseOptionsMultiple() {
     assertEquals("10.0", opts.get("max_distance"));
   }
 
+  @Test
+  void testMalformedOptionSegmentThrows() {
+    ExpressionEvaluationException ex =
+        assertThrows(
+            ExpressionEvaluationException.class,
+            () -> VectorSearchTableFunctionImplementation.parseOptions("k=5,badoption"));
+    assertTrue(ex.getMessage().contains("Malformed option segment"));
+  }
+
+  @Test
+  void testDuplicateOptionKeyThrows() {
+    ExpressionEvaluationException ex =
+        assertThrows(
+            ExpressionEvaluationException.class,
+            () -> VectorSearchTableFunctionImplementation.parseOptions("k=5,k=10"));
+    assertTrue(ex.getMessage().contains("Duplicate option key"));
+  }
+
+  @Test
+  void testEmptyVectorThrows() {
+    VectorSearchTableFunctionImplementation impl =
+        createImplWithArgs("my-index", "embedding", "[]", "k=5");
+    ExpressionEvaluationException ex =
+        assertThrows(ExpressionEvaluationException.class, () -> impl.applyArguments());
+    assertTrue(ex.getMessage().contains("must not be empty"));
+  }
+
+  @Test
+  void testMalformedVectorComponentThrows() {
+    VectorSearchTableFunctionImplementation impl =
+        createImplWithArgs("my-index", "embedding", "[1.0, abc, 3.0]", "k=5");
+    ExpressionEvaluationException ex =
+        assertThrows(ExpressionEvaluationException.class, () -> impl.applyArguments());
+    assertTrue(ex.getMessage().contains("Invalid vector component"));
+  }
+
+  @Test
+  void testNonFiniteVectorComponentThrows() {
+    VectorSearchTableFunctionImplementation impl =
+        createImplWithArgs("my-index", "embedding", "[1.0, Infinity, 3.0]", "k=5");
+    ExpressionEvaluationException ex =
+        assertThrows(ExpressionEvaluationException.class, () -> impl.applyArguments());
+    assertTrue(ex.getMessage().contains("must be a finite number"));
+  }
+
   @Test
   void testMissingArgumentThrows() {
     FunctionName functionName = FunctionName.of("vectorsearch");