Add vector search Phase 4: filter-aware ANN, SQL radius, HNSW efSearch, quantizers, IVF_ON_DISK, adaptive planner (#18119)

* Add vector search Phase 4: filter-aware ANN, SQL radius, HNSW efSearch, quantizers, IVF_ON_DISK, adaptive planner

This completes the vector search roadmap with seven feature areas:

1. Filter-aware ANN (FILTER_THEN_ANN): Pre-filter bitmap passed to HNSW/IVF
   backends for improved recall on selective filters
2. SQL surface: VECTOR_SIMILARITY_RADIUS for threshold/radius search
3. HNSW runtime tuning: vectorEfSearch query option via EfSearchAware interface
4. Generic quantizer framework: VectorQuantizerType (FLAT/SQ8/SQ4/PQ),
   ScalarQuantizer with train/encode/decode/serialize
5. IVF_ON_DISK: Disk-backed IVF via FileChannel reads (no 2GB limit)
6. Adaptive planner: VectorSearchStrategy with selectivity-aware mode selection
7. Metrics: VectorSearchMetrics singleton for observability

All existing configs, query options, and SQL are backward-compatible.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Fix threshold+exactRerank ordering bug, remove dead code, improve observability comments

- Fix CRITICAL: when both vectorExactRerank=true and vectorDistanceThreshold are set, the
  threshold-only branch fired first and returned early, bypassing exact rerank entirely.
  Now exact rerank takes priority and applies the threshold during the rerank step.
- Remove dead code: second threshold block (lines 382-390) that was unreachable because
  the first threshold branch always returned before it was reached.
- Remove duplicate setNumDocsMatchingAfterFilter call in record() (was called twice).
- Add LOGGER.warn when vectorEfSearch is set, making it visible that the option currently
  only affects EXPLAIN output and does not change Lucene graph traversal.
- Update VectorSimilarityFilterOperator Javadoc to list all 4 backends (was HNSW+IVF_FLAT only).
- Add comment in FilterPlanNode explaining why backendType/searchParams are null at the
  pre-filter wiring stage.
- Add comment in VectorIndexType noting IVF_ON_DISK reuses the IVF_FLAT file extension.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: xiangfu0

pinot-common/src/main/java/org/apache/pinot/common/request/context/RequestContextUtils.java

@@ -38,6 +38,7 @@
 import org.apache.pinot.common.request.context.predicate.RegexpLikePredicate;
 import org.apache.pinot.common.request.context.predicate.TextMatchPredicate;
 import org.apache.pinot.common.request.context.predicate.VectorSimilarityPredicate;
+import org.apache.pinot.common.request.context.predicate.VectorSimilarityRadiusPredicate;
 import org.apache.pinot.common.utils.RegexpPatternConverterUtils;
 import org.apache.pinot.common.utils.request.RequestUtils;
 import org.apache.pinot.segment.spi.AggregationFunctionType;
@@ -268,6 +269,15 @@ private static FilterContext getFilterInner(Function thriftFunction) {
           topK = operands.get(2).getLiteral().getIntValue();
         }
         return FilterContext.forPredicate(new VectorSimilarityPredicate(lhs, vectorValue, topK));
+      case VECTOR_SIMILARITY_RADIUS:
+        ExpressionContext radiusLhs = getExpression(operands.get(0));
+        float[] radiusVectorValue = getVectorValue(operands.get(1));
+        float threshold = VectorSimilarityRadiusPredicate.DEFAULT_THRESHOLD;
+        if (operands.size() == 3) {
+          threshold = getFloatValue(operands.get(2));
+        }
+        return FilterContext.forPredicate(
+            new VectorSimilarityRadiusPredicate(radiusLhs, radiusVectorValue, threshold));
       case IS_NULL:
         return FilterContext.forPredicate(new IsNullPredicate(getExpression(operands.get(0))));
       case IS_NOT_NULL:
@@ -449,6 +459,13 @@ private static FilterContext getFilterInner(FunctionContext filterFunction) {
         }
         return FilterContext.forPredicate(
             new VectorSimilarityPredicate(operands.get(0), getVectorValue(operands.get(1)), topK));
+      case VECTOR_SIMILARITY_RADIUS:
+        float radiusThreshold = VectorSimilarityRadiusPredicate.DEFAULT_THRESHOLD;
+        if (operands.size() == 3) {
+          radiusThreshold = Float.parseFloat(operands.get(2).getLiteral().getValue().toString());
+        }
+        return FilterContext.forPredicate(
+            new VectorSimilarityRadiusPredicate(operands.get(0), getVectorValue(operands.get(1)), radiusThreshold));
       case IS_NULL:
         return FilterContext.forPredicate(new IsNullPredicate(operands.get(0)));
       case IS_NOT_NULL:

pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/Predicate.java

@@ -38,7 +38,8 @@ enum Type {
     JSON_MATCH,
     IS_NULL,
     IS_NOT_NULL(true),
-    VECTOR_SIMILARITY;
+    VECTOR_SIMILARITY,
+    VECTOR_SIMILARITY_RADIUS;
 
     private final boolean _exclusive;
 

pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/VectorSimilarityRadiusPredicate.java

@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.request.context.predicate;
+
+import java.util.Arrays;
+import java.util.Objects;
+import org.apache.pinot.common.request.context.ExpressionContext;
+
+
+/**
+ * Predicate for vector similarity radius/threshold search.
+ * Returns all documents whose vector distance from the query vector is within the specified threshold.
+ *
+ * <p>Unlike {@link VectorSimilarityPredicate} which returns a fixed number of top-K results,
+ * this predicate returns all documents that satisfy the distance threshold constraint.</p>
+ *
+ * <p>An internal safety limit caps the maximum number of candidates retrieved from the ANN index
+ * before exact distance filtering is applied.</p>
+ *
+ * <p>Example SQL:</p>
+ * <pre>
+ *   WHERE VECTOR_SIMILARITY_RADIUS(embedding, ARRAY[1.0, 2.0, 3.0], 0.5)
+ * </pre>
+ *
+ * <p>NOTE: Currently, we only support vector similarity search on float array columns.</p>
+ */
+public class VectorSimilarityRadiusPredicate extends BasePredicate {
+  public static final float DEFAULT_THRESHOLD = 0.5f;
+  /**
+   * Default internal candidate limit for index-assisted radius search. This is used as a cap
+   * when retrieving ANN candidates before exact distance filtering. The effective limit is
+   * {@code Math.min(DEFAULT_INTERNAL_LIMIT, numDocs)} so it adapts to small segments.
+   * For large segments, consider increasing this via a query option if results appear truncated.
+   */
+  public static final int DEFAULT_INTERNAL_LIMIT = 100_000;
+
+  private final float[] _value;
+  private final float _threshold;
+
+  public VectorSimilarityRadiusPredicate(ExpressionContext lhs, float[] value, float threshold) {
+    super(lhs);
+    _value = value;
+    _threshold = threshold;
+  }
+
+  @Override
+  public Type getType() {
+    return Type.VECTOR_SIMILARITY_RADIUS;
+  }
+
+  public float[] getValue() {
+    return _value;
+  }
+
+  public float getThreshold() {
+    return _threshold;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof VectorSimilarityRadiusPredicate)) {
+      return false;
+    }
+    VectorSimilarityRadiusPredicate that = (VectorSimilarityRadiusPredicate) o;
+    return Objects.equals(_lhs, that._lhs) && Arrays.equals(_value, that._value)
+        && Float.compare(_threshold, that._threshold) == 0;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(_lhs, Arrays.hashCode(_value), _threshold);
+  }
+
+  @Override
+  public String toString() {
+    return "vector_similarity_radius(" + _lhs + ",'" + Arrays.toString(_value) + "'," + _threshold + ")";
+  }
+}

pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java

@@ -729,6 +729,16 @@ public static Float getVectorDistanceThreshold(Map<String, String> queryOptions)
     }
   }
 
+  /**
+   * Returns the configured efSearch value for HNSW vector search, or {@code null} if not set.
+   */
+  @Nullable
+  public static Integer getVectorEfSearch(Map<String, String> queryOptions) {
+    String efSearch = queryOptions.get(QueryOptionKey.VECTOR_EF_SEARCH);
+    return checkedParseIntPositive(QueryOptionKey.VECTOR_EF_SEARCH, efSearch);
+  }
+
+
   public static int getSortExchangeCopyThreshold(Map<String, String> options, int i) {
     String sortExchangeCopyThreshold = options.get(QueryOptionKey.SORT_EXCHANGE_COPY_THRESHOLD);
     if (sortExchangeCopyThreshold != null) {

pinot-common/src/main/java/org/apache/pinot/sql/FilterKind.java

@@ -38,7 +38,8 @@ public enum FilterKind {
   JSON_MATCH,
   IS_NULL,
   IS_NOT_NULL,
-  VECTOR_SIMILARITY;
+  VECTOR_SIMILARITY,
+  VECTOR_SIMILARITY_RADIUS;
 
   /**
    * Helper method that returns true if the enum maps to a Range.

pinot-common/src/main/java/org/apache/pinot/sql/parsers/CalciteSqlParser.java

@@ -299,6 +299,30 @@ private static void validateFilter(Expression filterExpression) {
               + "the signature is VECTOR_SIMILARITY(float[], float[], int)");
         }
       }
+    } else if (operator.equals(FilterKind.VECTOR_SIMILARITY_RADIUS.name())) {
+      Expression vectorIdentifier = filterExpression.getFunctionCall().getOperands().get(0);
+      if (!vectorIdentifier.isSetIdentifier()) {
+        throw new IllegalStateException(
+            "The first argument of VECTOR_SIMILARITY_RADIUS must be an identifier of float array, "
+                + "the signature is VECTOR_SIMILARITY_RADIUS(float[], float[], float).");
+      }
+      Expression vectorLiteral = filterExpression.getFunctionCall().getOperands().get(1);
+      if ((vectorLiteral.isSetFunctionCall() && !vectorLiteral.getFunctionCall().getOperator().equalsIgnoreCase(
+          "arrayvalueconstructor"))
+          || (vectorLiteral.isSetLiteral() && !vectorLiteral.getLiteral().isSetFloatArrayValue()
+          && !vectorLiteral.getLiteral().isSetDoubleArrayValue())) {
+        throw new IllegalStateException(
+            "The second argument of VECTOR_SIMILARITY_RADIUS must be a float/double array "
+                + "literal, the signature is VECTOR_SIMILARITY_RADIUS(float[], float[], float)");
+      }
+      if (filterExpression.getFunctionCall().getOperands().size() == 3) {
+        Expression threshold = filterExpression.getFunctionCall().getOperands().get(2);
+        if (!threshold.isSetLiteral()) {
+          throw new IllegalStateException(
+              "The third argument of VECTOR_SIMILARITY_RADIUS must be a numeric literal, "
+                  + "the signature is VECTOR_SIMILARITY_RADIUS(float[], float[], float)");
+        }
+      }
     } else {
       List<Expression> operands = filterExpression.getFunctionCall().getOperands();
       for (int i = 1; i < operands.size(); i++) {

pinot-common/src/main/java/org/apache/pinot/sql/parsers/rewriter/PredicateComparisonRewriter.java

@@ -148,6 +148,25 @@ private static Expression updateFunctionExpression(Expression expression) {
           }
           break;
         }
+        case VECTOR_SIMILARITY_RADIUS: {
+          Preconditions.checkArgument(operands.size() >= 2 && operands.size() <= 3,
+              "For %s predicate, the number of operands must be at either 2 or 3, got: %s", filterKind, expression);
+          if ((operands.get(1).getFunctionCall() != null && !operands.get(1).getFunctionCall().getOperator()
+              .equalsIgnoreCase("arrayvalueconstructor"))
+              || (operands.get(1).getLiteral() != null && !operands.get(1).getLiteral().isSetFloatArrayValue()
+                  && !operands.get(1).getLiteral().isSetDoubleArrayValue())) {
+            throw new SqlCompilationException(
+                String.format("For %s predicate, the second operand must be a float/double array literal, got: %s",
+                    filterKind,
+                    expression));
+          }
+          if (operands.size() == 3 && operands.get(2).getLiteral() == null) {
+            throw new SqlCompilationException(
+                String.format("For %s predicate, the third operand must be a literal, got: %s", filterKind,
+                    expression));
+          }
+          break;
+        }
         default:
           int numOperands = operands.size();
           for (int i = 1; i < numOperands; i++) {

pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/VectorSimilarityRadiusPredicateTest.java

@@ -0,0 +1,127 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.request.context.predicate;
+
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * Tests for {@link VectorSimilarityRadiusPredicate}.
+ */
+public class VectorSimilarityRadiusPredicateTest {
+
+  @Test
+  public void testConstructionAndGetters() {
+    ExpressionContext lhs = ExpressionContext.forIdentifier("embedding");
+    float[] value = {1.0f, 2.0f, 3.0f};
+    float threshold = 0.75f;
+
+    VectorSimilarityRadiusPredicate predicate = new VectorSimilarityRadiusPredicate(lhs, value, threshold);
+
+    Assert.assertEquals(predicate.getType(), Predicate.Type.VECTOR_SIMILARITY_RADIUS);
+    Assert.assertEquals(predicate.getLhs(), lhs);
+    Assert.assertEquals(predicate.getValue(), value);
+    Assert.assertEquals(predicate.getThreshold(), threshold);
+  }
+
+  @Test
+  public void testDefaultThreshold() {
+    Assert.assertEquals(VectorSimilarityRadiusPredicate.DEFAULT_THRESHOLD, 0.5f);
+  }
+
+  @Test
+  public void testDefaultInternalLimit() {
+    Assert.assertEquals(VectorSimilarityRadiusPredicate.DEFAULT_INTERNAL_LIMIT, 100_000);
+  }
+
+  @Test
+  public void testEquality() {
+    ExpressionContext lhs1 = ExpressionContext.forIdentifier("embedding");
+    ExpressionContext lhs2 = ExpressionContext.forIdentifier("embedding");
+    float[] value1 = {1.0f, 2.0f};
+    float[] value2 = {1.0f, 2.0f};
+
+    VectorSimilarityRadiusPredicate p1 = new VectorSimilarityRadiusPredicate(lhs1, value1, 0.5f);
+    VectorSimilarityRadiusPredicate p2 = new VectorSimilarityRadiusPredicate(lhs2, value2, 0.5f);
+
+    Assert.assertEquals(p1, p2);
+    Assert.assertEquals(p1.hashCode(), p2.hashCode());
+  }
+
+  @Test
+  public void testInequalityDifferentThreshold() {
+    ExpressionContext lhs = ExpressionContext.forIdentifier("embedding");
+    float[] value = {1.0f, 2.0f};
+
+    VectorSimilarityRadiusPredicate p1 = new VectorSimilarityRadiusPredicate(lhs, value, 0.5f);
+    VectorSimilarityRadiusPredicate p2 = new VectorSimilarityRadiusPredicate(lhs, value, 1.0f);
+
+    Assert.assertNotEquals(p1, p2);
+  }
+
+  @Test
+  public void testInequalityDifferentVector() {
+    ExpressionContext lhs = ExpressionContext.forIdentifier("embedding");
+    float[] value1 = {1.0f, 2.0f};
+    float[] value2 = {3.0f, 4.0f};
+
+    VectorSimilarityRadiusPredicate p1 = new VectorSimilarityRadiusPredicate(lhs, value1, 0.5f);
+    VectorSimilarityRadiusPredicate p2 = new VectorSimilarityRadiusPredicate(lhs, value2, 0.5f);
+
+    Assert.assertNotEquals(p1, p2);
+  }
+
+  @Test
+  public void testInequalityDifferentColumn() {
+    ExpressionContext lhs1 = ExpressionContext.forIdentifier("embedding1");
+    ExpressionContext lhs2 = ExpressionContext.forIdentifier("embedding2");
+    float[] value = {1.0f, 2.0f};
+
+    VectorSimilarityRadiusPredicate p1 = new VectorSimilarityRadiusPredicate(lhs1, value, 0.5f);
+    VectorSimilarityRadiusPredicate p2 = new VectorSimilarityRadiusPredicate(lhs2, value, 0.5f);
+
+    Assert.assertNotEquals(p1, p2);
+  }
+
+  @Test
+  public void testNotEqualToVectorSimilarityPredicate() {
+    ExpressionContext lhs = ExpressionContext.forIdentifier("embedding");
+    float[] value = {1.0f, 2.0f};
+
+    VectorSimilarityRadiusPredicate radiusPredicate = new VectorSimilarityRadiusPredicate(lhs, value, 0.5f);
+    VectorSimilarityPredicate topKPredicate = new VectorSimilarityPredicate(lhs, value, 10);
+
+    Assert.assertNotEquals(radiusPredicate, topKPredicate);
+  }
+
+  @Test
+  public void testToString() {
+    ExpressionContext lhs = ExpressionContext.forIdentifier("embedding");
+    float[] value = {1.0f, 2.0f};
+
+    VectorSimilarityRadiusPredicate predicate = new VectorSimilarityRadiusPredicate(lhs, value, 0.5f);
+    String str = predicate.toString();
+
+    Assert.assertTrue(str.contains("vector_similarity_radius"));
+    Assert.assertTrue(str.contains("embedding"));
+    Assert.assertTrue(str.contains("0.5"));
+  }
+}