This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch PHOENIX-7998 in repository https://gitbox.apache.org/repos/asf/phoenix.git
commit 9d9e6b82ec01262672d5845e36dd6d3f3192beb8 Author: Andrew Purtell <[email protected]> AuthorDate: Tue Oct 6 00:15:12 2026 -0700 PHOENIX-7998 Phase 12: Adaptive probing and hybrid filtering Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> --- .../java/org/apache/phoenix/execute/ScanPlan.java | 20 +- .../phoenix/execute/VectorIndexScanPlan.java | 269 ++++++++++++- .../apache/phoenix/optimize/OptimizerReasons.java | 1 + .../apache/phoenix/optimize/QueryOptimizer.java | 67 ++-- .../apache/phoenix/optimize/VectorSearchUtil.java | 56 ++- .../java/org/apache/phoenix/parse/HintNode.java | 7 +- .../org/apache/phoenix/query/QueryServices.java | 2 + .../apache/phoenix/query/QueryServicesOptions.java | 1 + .../phoenix/end2end/VectorHybridFilterIT.java | 416 +++++++++++++++++++++ .../phoenix/execute/VectorIndexScanPlanTest.java | 3 +- .../phoenix/optimize/VectorPlanOrderingTest.java | 45 +++ 11 files changed, 850 insertions(+), 37 deletions(-) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/execute/ScanPlan.java b/phoenix-core-client/src/main/java/org/apache/phoenix/execute/ScanPlan.java index a98f2e80fc..5cc227e737 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/execute/ScanPlan.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/execute/ScanPlan.java @@ -90,7 +90,7 @@ public class ScanPlan extends BaseQueryPlan { private List<KeyRange> splits; private List<List<Scan>> scans; private boolean allowPageFilter; - private boolean isSerial; + protected boolean isSerial; private boolean isDataToScanWithinThreshold; private Long serialRowsEstimate; private Long serialBytesEstimate; @@ -273,6 +273,18 @@ public class ScanPlan extends BaseQueryPlan { && ScanUtil.shouldRowsBeInRowKeyOrder(orderBy, context)); } + /** + * Populates plan scan estimates, splits, and parallel scan metadata from the underlying result + * iterators. + */ + protected void recordIteratorStats(BaseResultIterators iterators) { + estimatedRows = iterators.getEstimatedRowCount(); + estimatedSize = iterators.getEstimatedByteCount(); + estimateInfoTimestamp = iterators.getEstimateInfoTimestamp(); + splits = iterators.getSplits(); + scans = iterators.getScans(); + } + @Override protected ResultIterator newIterator(ParallelScanGrouper scanGrouper, Scan scan, Map<ImmutableBytesPtr, ServerCache> caches) throws SQLException { @@ -310,11 +322,7 @@ public class ScanPlan extends BaseQueryPlan { iterators = new ParallelIterators(this, perScanLimit, parallelIteratorFactory, scanGrouper, scan, initFirstScanOnly, caches, dataPlan); } - estimatedRows = iterators.getEstimatedRowCount(); - estimatedSize = iterators.getEstimatedByteCount(); - estimateInfoTimestamp = iterators.getEstimateInfoTimestamp(); - splits = iterators.getSplits(); - scans = iterators.getScans(); + recordIteratorStats(iterators); if (isOffsetOnServer) { scanner = new ConcatResultIterator(iterators); if (limit != null) { diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/execute/VectorIndexScanPlan.java b/phoenix-core-client/src/main/java/org/apache/phoenix/execute/VectorIndexScanPlan.java index 4ceb71ee0a..1d4da8bfd0 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/execute/VectorIndexScanPlan.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/execute/VectorIndexScanPlan.java @@ -17,11 +17,19 @@ */ package org.apache.phoenix.execute; +import java.io.IOException; import java.sql.SQLException; import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.filter.Filter; +import org.apache.hadoop.hbase.filter.FilterList; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.phoenix.cache.ServerCacheClient.ServerCache; import org.apache.phoenix.cache.VectorCentroidCache; import org.apache.phoenix.cache.VectorCentroidCache.CachedCentroids; import org.apache.phoenix.compile.ExplainPlan; @@ -32,9 +40,19 @@ import org.apache.phoenix.compile.QueryPlan; import org.apache.phoenix.compile.RowProjector; import org.apache.phoenix.compile.ScanRanges; import org.apache.phoenix.compile.StatementContext; +import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants; import org.apache.phoenix.expression.Expression; +import org.apache.phoenix.expression.OrderByExpression; import org.apache.phoenix.expression.function.DistanceFunction; +import org.apache.phoenix.filter.SkipScanFilter; +import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; +import org.apache.phoenix.iterate.BaseResultIterators; +import org.apache.phoenix.iterate.MergeSortTopNResultIterator; import org.apache.phoenix.iterate.ParallelIteratorFactory; +import org.apache.phoenix.iterate.ParallelIterators; +import org.apache.phoenix.iterate.ParallelScanGrouper; +import org.apache.phoenix.iterate.ResultIterator; +import org.apache.phoenix.iterate.SerialIterators; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.optimize.Cost; import org.apache.phoenix.optimize.DistanceMetric; @@ -42,12 +60,15 @@ import org.apache.phoenix.parse.FilterableStatement; import org.apache.phoenix.parse.HintNode; import org.apache.phoenix.parse.HintNode.Hint; import org.apache.phoenix.query.KeyRange; +import org.apache.phoenix.query.QueryConstants; import org.apache.phoenix.query.QueryServices; import org.apache.phoenix.query.QueryServicesOptions; import org.apache.phoenix.schema.PTable; import org.apache.phoenix.schema.SaltingUtil; import org.apache.phoenix.schema.TableRef; +import org.apache.phoenix.schema.tuple.Tuple; import org.apache.phoenix.schema.types.PInteger; +import org.apache.phoenix.util.ClientUtil; import org.apache.phoenix.util.CostUtil; import org.apache.phoenix.util.ReadOnlyProps; import org.apache.phoenix.util.ScanUtil; @@ -69,6 +90,11 @@ public class VectorIndexScanPlan extends ScanPlan { private final int centroidCount; private final int probeCount; private final int[] probeCentroids; + private final int maxProbeBatches; + private final CachedCentroids centroids; + private final float[] queryVector; + // Active posting list skip scan filter, updated across successive adaptive probing batches + private final SkipScanFilter postingListFilter; public VectorIndexScanPlan(StatementContext context, FilterableStatement statement, TableRef table, RowProjector projector, Integer limit, Integer offset, OrderBy orderBy, @@ -81,15 +107,17 @@ public class VectorIndexScanPlan extends ScanPlan { ReadOnlyProps props = connection.getQueryServices().getProps(); HintNode hint = statement.getHint(); - float[] queryVector = getQueryVector(orderBy); + this.queryVector = getQueryVector(orderBy); CachedCentroids centroids = null; if (queryVector != null && index.getVectorCentroidGeneration() != null) { centroids = VectorCentroidCache.getInstance(connection.getQueryServices().getConfiguration()) .get(connection, index.getName().getString(), index.getVectorCentroidGeneration(), DistanceMetric.fromString(index.getVectorDistanceMetric())); } + this.centroids = centroids; this.centroidCount = centroids == null ? 0 : centroids.size(); int[] probes = new int[0]; + SkipScanFilter filter = null; if (centroids != null) { int requested = getHintInt(hint, Hint.VECTOR_PROBE_COUNT, props.getInt( QueryServices.VECTOR_PROBE_COUNT_ATTRIB, QueryServicesOptions.DEFAULT_VECTOR_PROBE_COUNT)); @@ -102,12 +130,17 @@ public class VectorIndexScanPlan extends ScanPlan { probes = nearest; context.setScanRanges(ranges); if (ranges.useSkipScanFilter()) { - ScanUtil.andFilterAtBeginning(context.getScan(), ranges.getSkipScanFilter()); + filter = ranges.getSkipScanFilter(); + ScanUtil.andFilterAtBeginning(context.getScan(), filter); } } } this.probeCentroids = probes; this.probeCount = probes.length; + this.postingListFilter = filter; + this.maxProbeBatches = getHintInt(hint, Hint.MAX_PROBE_LIMIT, + props.getInt(QueryServices.VECTOR_MAX_PROBE_LIMIT_ATTRIB, + QueryServicesOptions.DEFAULT_VECTOR_MAX_PROBE_LIMIT)); } /** Extracts the constant query vector from the ORDER BY distance expression if present. */ @@ -154,7 +187,18 @@ public class VectorIndexScanPlan extends ScanPlan { if (!compiled.isEverything() && compiled.getBoundPkColumnCount() > prefixSlots) { return null; } - List<List<KeyRange>> slots = new ArrayList<>(prefixSlots + 1); + return buildPostingListRanges(context, index, centroids); + } + + /** + * Constructs scan ranges spanning candidate centroid posting lists across tenant and salting key + * prefixes. + */ + private static ScanRanges buildPostingListRanges(StatementContext context, PTable index, + int[] centroids) throws SQLException { + boolean isSalted = index.getBucketNum() != null; + boolean scopeTenant = index.isMultiTenant(); + List<List<KeyRange>> slots = new ArrayList<>(3); if (isSalted) { slots.add(SaltingUtil.generateAllSaltingRanges(index.getBucketNum())); } @@ -235,8 +279,13 @@ public class VectorIndexScanPlan extends ScanPlan { } String metric = getTableRef().getTable().getVectorDistanceMetric(); List<String> steps = new ArrayList<>(explainPlan.getPlanSteps()); - steps.add(0, - "CLIENT PROBING " + probeCount + " OF " + centroidCount + " CENTROIDS (" + metric + ")"); + String probing = + "CLIENT PROBING " + probeCount + " OF " + centroidCount + " CENTROIDS (" + metric + ")"; + if (isAdaptive()) { + int batches = Math.min(maxProbeBatches, (centroidCount + probeCount - 1) / probeCount); + probing += " EXPANDING UP TO " + batches + " BATCHES"; + } + steps.add(0, probing); ExplainPlanAttributes attributes = explainPlan.getPlanStepsAsAttributes(); return new ExplainPlan(steps, (attributes == null @@ -245,7 +294,215 @@ public class VectorIndexScanPlan extends ScanPlan { .setVectorCentroidCount(centroidCount).setVectorDistanceMetric(metric).build()); } - /** Number of probed centroid posting lists, or 0 for full index scans. */ + /** + * Indicates whether query execution requires adaptive batch expansion over centroid posting lists + * due to non-exhaustive initial probing and filter selectivity. + */ + public boolean isAdaptive() { + return limit != null && probeCount > 0 && probeCount < centroidCount + && statement.getWhere() != null && maxProbeBatches > 1; + } + + @Override + protected ResultIterator newIterator(ParallelScanGrouper scanGrouper, Scan scan, + Map<ImmutableBytesPtr, ServerCache> caches) throws SQLException { + if (!isAdaptive()) { + return super.newIterator(scanGrouper, scan, caches); + } + scan.setAttribute(BaseScannerRegionObserverConstants.NON_AGGREGATE_QUERY, QueryConstants.TRUE); + // Initialize initial probe batch iterators to capture baseline plan statistics and splits + BaseResultIterators first = newBatch(scanGrouper, scan, caches); + recordIteratorStats(first); + return new AdaptiveProbeResultIterator(scanGrouper, scan, caches, first); + } + + private BaseResultIterators newBatch(ParallelScanGrouper scanGrouper, Scan scan, + Map<ImmutableBytesPtr, ServerCache> caches) throws SQLException { + return isSerial + ? new SerialIterators(this, null, null, parallelIteratorFactory, scanGrouper, scan, caches, + dataPlan) + : new ParallelIterators(this, null, parallelIteratorFactory, scanGrouper, scan, false, caches, + dataPlan); + } + + /** + * Replaces the posting list filter within the scan filter tree with the target batch filter, + * preserving filter structure and position. + */ + private Filter replacePostingListFilter(Filter filter, SkipScanFilter next) { + if (filter == postingListFilter) { + return next; + } + if (filter instanceof FilterList) { + List<Filter> filters = new ArrayList<>(((FilterList) filter).getFilters()); + int i = filters.indexOf(postingListFilter); + if (i >= 0) { + if (next == null) { + filters.remove(i); + } else { + filters.set(i, next); + } + return new FilterList(((FilterList) filter).getOperator(), filters); + } + } + if (next == null) { + return filter; + } + return filter == null ? next : new FilterList(FilterList.Operator.MUST_PASS_ALL, next, filter); + } + + /** + * Result iterator executing adaptive multi-batch scans across successive centroid posting lists, + * aggregating and globally reranking candidate tuples to satisfy filter and limit criteria. + */ + private class AdaptiveProbeResultIterator implements ResultIterator { + private final ParallelScanGrouper scanGrouper; + private final Scan scan; + private final Map<ImmutableBytesPtr, ServerCache> caches; + private final BaseResultIterators first; + private final int batchLimit; + private List<Tuple> rows; + private int next; + + AdaptiveProbeResultIterator(ParallelScanGrouper scanGrouper, Scan scan, + Map<ImmutableBytesPtr, ServerCache> caches, BaseResultIterators first) { + this.scanGrouper = scanGrouper; + this.scan = scan; + this.caches = caches; + this.first = first; + this.batchLimit = limit + (offset == null ? 0 : offset); + } + + private List<OrderByExpression> orderByExpressions() { + return orderBy.getOrderByExpressions(); + } + + private void drain(ResultIterator batch, List<Tuple> into) throws SQLException { + try { + for (Tuple t = batch.next(); t != null; t = batch.next()) { + into.add(t); + } + } finally { + batch.close(); + } + } + + private void probe() throws SQLException { + List<Tuple> found = new ArrayList<>(); + drain(new MergeSortTopNResultIterator(first, batchLimit, null, orderByExpressions()), found); + Set<Integer> probed = new HashSet<>(); + for (int c : probeCentroids) { + probed.add(c); + } + int[] ranked = null; + int batches = 1; + ScanRanges original = context.getScanRanges(); + try { + while ( + found.size() < batchLimit && probed.size() < centroidCount && batches < maxProbeBatches + ) { + if (ranked == null) { + ranked = centroids.nearest(queryVector, centroidCount); + } + List<Integer> batch = new ArrayList<>(probeCount); + for (int i = 0; i < ranked.length && batch.size() < probeCount; i++) { + if (probed.add(ranked[i])) { + batch.add(ranked[i]); + } + } + int[] ids = new int[batch.size()]; + for (int i = 0; i < ids.length; i++) { + ids[i] = batch.get(i); + } + ScanRanges ranges = buildPostingListRanges(context, getTableRef().getTable(), ids); + Scan batchScan; + try { + batchScan = new Scan(scan); + } catch (IOException e) { + throw ClientUtil.parseServerException(e); + } + batchScan.setFilter(replacePostingListFilter(scan.getFilter(), + ranges.useSkipScanFilter() ? ranges.getSkipScanFilter() : null)); + ranges.initializeScan(batchScan); + // Synchronize context scan ranges for subsequent iterator construction + context.setScanRanges(ranges); + drain(new MergeSortTopNResultIterator(newBatch(scanGrouper, batchScan, caches), + batchLimit, null, orderByExpressions()), found); + batches++; + } + } finally { + context.setScanRanges(original); + } + found.sort(new TupleComparator(orderByExpressions())); + int from = Math.min(offset == null ? 0 : offset, found.size()); + rows = found.subList(from, Math.min(from + limit, found.size())); + } + + @Override + public Tuple next() throws SQLException { + if (rows == null) { + probe(); + } + return next < rows.size() ? rows.get(next++) : null; + } + + @Override + public void close() throws SQLException { + if (rows == null) { + first.close(); + } + } + + @Override + public void explain(List<String> planSteps) { + new MergeSortTopNResultIterator(first, limit, offset, orderByExpressions()) + .explain(planSteps); + } + + @Override + public void explain(List<String> planSteps, + ExplainPlanAttributesBuilder explainPlanAttributesBuilder) { + new MergeSortTopNResultIterator(first, limit, offset, orderByExpressions()).explain(planSteps, + explainPlanAttributesBuilder); + } + } + + /** Comparator enforcing client side tuple ordering matching compiled ORDER BY expressions. */ + private static final class TupleComparator implements java.util.Comparator<Tuple> { + private final List<OrderByExpression> orderBy; + private final ImmutableBytesWritable ptr1 = new ImmutableBytesWritable(); + private final ImmutableBytesWritable ptr2 = new ImmutableBytesWritable(); + + TupleComparator(List<OrderByExpression> orderBy) { + this.orderBy = orderBy; + } + + @Override + public int compare(Tuple t1, Tuple t2) { + for (OrderByExpression order : orderBy) { + boolean isNull1 = !order.getExpression().evaluate(t1, ptr1) || ptr1.getLength() == 0; + boolean isNull2 = !order.getExpression().evaluate(t2, ptr2) || ptr2.getLength() == 0; + if (isNull1 || isNull2) { + if (isNull1 != isNull2) { + return isNull1 == order.isNullsLast() ? 1 : -1; + } + continue; + } + int c = ptr1.compareTo(ptr2); + if (c != 0) { + return order.isAscending() ? c : -c; + } + } + return 0; + } + } + + /** Maximum number of probe batches permitted during adaptive centroid expansion. */ + public int getMaxProbeBatches() { + return maxProbeBatches; + } + + /** The number of posting lists the plan scans, or 0 when it scans the whole index. */ public int getProbeCount() { return probeCount; } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/OptimizerReasons.java b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/OptimizerReasons.java index 5ee0235613..0426f9cd21 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/OptimizerReasons.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/OptimizerReasons.java @@ -36,6 +36,7 @@ public final class OptimizerReasons { public static final String RULE_NON_LOCAL_PREFERRED = "non-local preferred"; public static final String RULE_PARTIAL_INDEX_APPLICABLE = "partial index applicable"; public static final String RULE_NEAREST_NEIGHBOR_INDEX = "nearest-neighbor index"; + public static final String RULE_SELECTIVE_FILTER_FIRST = "selective filter first"; // REASON_* — rejected-index reason labels. public static final String REASON_NO_PK_PREFIX_BOUND = "no PK prefix bound"; diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java index 9a8f2b147a..3e42be2684 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java @@ -915,13 +915,16 @@ public class QueryOptimizer { final boolean useDataOverIndexHint = select.getHint().hasHint(Hint.USE_DATA_OVER_INDEX_TABLE); final int comparisonOfDataVersusIndexTable = useDataOverIndexHint ? -1 : 1; + final Set<QueryPlan> filterFirst = VectorSearchUtil.getFilterFirstPlans(plans, dataPlan); if (this.costBased) { Collections.sort(plans, new Comparator<QueryPlan>() { @Override public int compare(QueryPlan plan1, QueryPlan plan2) { int c = VectorSearchUtil.getCost(plan1).compareTo(VectorSearchUtil.getCost(plan2)); // Break cost ties involving vector index plans using vector static precedence rules. - return c != 0 ? c : compareVectorPlans(plan1, plan2, comparisonOfDataVersusIndexTable); + return c != 0 + ? c + : compareVectorPlans(plan1, plan2, comparisonOfDataVersusIndexTable, filterFirst); } }); // Return ordered list based on cost if stats are available; otherwise fall @@ -993,7 +996,7 @@ public class QueryOptimizer { public int compare(QueryPlan plan1, QueryPlan plan2) { PTable table1 = plan1.getTableRef().getTable(); PTable table2 = plan2.getTableRef().getTable(); - int c = compareVectorPlans(plan1, plan2, comparisonOfDataVersusIndexTable); + int c = compareVectorPlans(plan1, plan2, comparisonOfDataVersusIndexTable, filterFirst); if (c != 0) return c; int boundCount1 = plan1.getContext().getScanRanges().getBoundPkColumnCount(); int boundCount2 = plan2.getContext().getScanRanges().getBoundPkColumnCount(); @@ -1063,30 +1066,44 @@ public class QueryOptimizer { QueryPlan runnerUp = bestCandidates.size() > 1 ? bestCandidates.get(1) : firstOther(plans, winner); tagComparatorRejections(winner, plans, boundRanges, state); - recordDecision(winner, labelComparatorRule(winner, runnerUp, boundRanges), state); + recordDecision(winner, labelComparatorRule(winner, runnerUp, boundRanges, filterFirst), state); return stopAtBestPlan ? bestCandidates.subList(0, 1) : bestCandidates; } + static int compareVectorPlans(QueryPlan plan1, QueryPlan plan2, + int comparisonOfDataVersusIndexTable) { + return compareVectorPlans(plan1, plan2, comparisonOfDataVersusIndexTable, + Collections.emptySet()); + } + /** - * Compares plans based on vector index precedence. - * <p> - * Prioritizes vector index plans over non-vector plans for nearest neighbor queries unless base - * table execution is hinted. Among competing vector index plans, preference is given to plans - * requiring fewer base table lookups (covering index, then deferred projection, then filter-time - * joins). Returns 0 when neither plan uses a vector index to preserve external ordering. + * Compares candidate query plans for nearest neighbor vector searches according to access tier + * and index coverage. Highly selective secondary index plans + * ({@link VectorSearchUtil#getFilterFirstPlans}) take precedence over vector index scans unless + * overridden by data table hints. Within vector index plans, preference is given to plans + * requiring fewer base table lookups (covering index, then deferred projection, then join scans). + * Returns 0 when neither plan utilizes vector indexing or selective filter first evaluation, + * preserving relative order and transitivity across tiers. */ static int compareVectorPlans(QueryPlan plan1, QueryPlan plan2, - int comparisonOfDataVersusIndexTable) { - boolean vector1 = VectorSearchUtil.usesVectorIndex(plan1); - boolean vector2 = VectorSearchUtil.usesVectorIndex(plan2); - if (vector1 != vector2) { - return vector1 ? -comparisonOfDataVersusIndexTable : comparisonOfDataVersusIndexTable; - } - return vector1 - ? Integer.compare(VectorSearchUtil.getLookupRank(plan1), - VectorSearchUtil.getLookupRank(plan2)) - : 0; + int comparisonOfDataVersusIndexTable, Set<QueryPlan> filterFirst) { + int c = Integer.compare(vectorTier(plan1, comparisonOfDataVersusIndexTable, filterFirst), + vectorTier(plan2, comparisonOfDataVersusIndexTable, filterFirst)); + if (c != 0 || !VectorSearchUtil.usesVectorIndex(plan1)) { + return c; + } + return Integer.compare(VectorSearchUtil.getLookupRank(plan1), + VectorSearchUtil.getLookupRank(plan2)); + } + + private static int vectorTier(QueryPlan plan, int comparisonOfDataVersusIndexTable, + Set<QueryPlan> filterFirst) { + boolean dataHinted = comparisonOfDataVersusIndexTable < 0; + if (VectorSearchUtil.usesVectorIndex(plan)) { + return dataHinted ? 3 : 1; + } + return filterFirst.contains(plan) && !dataHinted ? 0 : 2; } /** @@ -1274,12 +1291,18 @@ public class QueryOptimizer { * Returns the {@code RULE_*} label for the chosen plan by replaying the comparator ladder in * {@link #orderPlansBestToWorst} against the runner up and returning the first decisive branch. */ - private static String labelComparatorRule(QueryPlan winner, QueryPlan runnerUp, int boundRanges) { + private static String labelComparatorRule(QueryPlan winner, QueryPlan runnerUp, int boundRanges, + Set<QueryPlan> filterFirst) { if (runnerUp == null) { return OptimizerReasons.RULE_ONLY_CANDIDATE; } - if (VectorSearchUtil.usesVectorIndex(winner) && compareVectorPlans(winner, runnerUp, 1) != 0) { - return OptimizerReasons.RULE_NEAREST_NEIGHBOR_INDEX; + if (compareVectorPlans(winner, runnerUp, 1, filterFirst) != 0) { + if (VectorSearchUtil.usesVectorIndex(winner)) { + return OptimizerReasons.RULE_NEAREST_NEIGHBOR_INDEX; + } + if (filterFirst.contains(winner)) { + return OptimizerReasons.RULE_SELECTIVE_FILTER_FIRST; + } } PTable winnerTable = winner.getTableRef().getTable(); PTable runnerUpTable = runnerUp.getTableRef().getTable(); diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/VectorSearchUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/VectorSearchUtil.java index 614a42a4af..d9fd9ad2b3 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/VectorSearchUtil.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/VectorSearchUtil.java @@ -18,7 +18,11 @@ package org.apache.phoenix.optimize; import java.lang.reflect.Array; +import java.sql.SQLException; +import java.util.Collections; +import java.util.IdentityHashMap; import java.util.List; +import java.util.Set; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.phoenix.compile.OrderByCompiler.OrderBy; import org.apache.phoenix.compile.QueryPlan; @@ -38,6 +42,7 @@ import org.apache.phoenix.parse.LiteralParseNode; import org.apache.phoenix.parse.OrderByNode; import org.apache.phoenix.parse.ParseNode; import org.apache.phoenix.parse.SelectStatement; +import org.apache.phoenix.schema.PTableType; import org.apache.phoenix.schema.types.PDataType; import org.apache.phoenix.schema.types.PVectorDataType; import org.apache.phoenix.util.CostUtil; @@ -238,7 +243,56 @@ public final class VectorSearchUtil { : null; } - /** Indicates whether the query execution plan incorporates a vector index scan. */ + /** + * Maximum table row scan ratio threshold for evaluating selective secondary index filters ahead + * of approximate vector index probing. + */ + public static final double FILTER_FIRST_MAX_ROW_FRACTION = 0.05; + + /** + * Identifies candidate secondary index plans whose estimated row scan counts fall within the + * selectivity threshold relative to the primary data table. When statistics demonstrate high + * filter selectivity, exact post-filtering over the secondary index is preferred to vector index + * probing. If statistics are unavailable or no vector index plan exists, returns an empty set. + */ + public static Set<QueryPlan> getFilterFirstPlans(List<QueryPlan> plans, QueryPlan dataPlan) { + Set<QueryPlan> filterFirst = Collections.newSetFromMap(new IdentityHashMap<>()); + boolean hasVectorPlan = false; + for (QueryPlan plan : plans) { + hasVectorPlan |= usesVectorIndex(plan); + } + if (!hasVectorPlan) { + return filterFirst; + } + Long dataRows = null; + try { + dataRows = dataPlan.getEstimatedRowsToScan(); + } catch (SQLException e) { + // Missing cardinality estimate disables filter first selection + } + if (dataRows == null || dataRows <= 0) { + return filterFirst; + } + for (QueryPlan plan : plans) { + if ( + plan == dataPlan || usesVectorIndex(plan) + || plan.getTableRef().getTable().getType() != PTableType.INDEX + ) { + continue; + } + try { + Long rows = plan.getEstimatedRowsToScan(); + if (rows != null && rows <= dataRows * FILTER_FIRST_MAX_ROW_FRACTION) { + filterFirst.add(plan); + } + } catch (SQLException e) { + // Unestimated plan disqualified from filter first selection + } + } + return filterFirst; + } + + /** Returns true when the plan reads a vector index; see {@link #getVectorIndexScan}. */ public static boolean usesVectorIndex(QueryPlan plan) { return getVectorIndexScan(plan) != null; } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/HintNode.java b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/HintNode.java index ae53cfae9c..b161989ba3 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/HintNode.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/HintNode.java @@ -130,7 +130,12 @@ public class HintNode { /** * Overrides the number of IVF centroids probed during vector index scans. */ - VECTOR_PROBE_COUNT; + VECTOR_PROBE_COUNT, + + /** + * Overrides the maximum number of probe batches scanned during adaptive vector index search. + */ + MAX_PROBE_LIMIT; }; private final Map<Hint, String> hints; diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java index 93b31b36e8..3a43a518cb 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java @@ -154,6 +154,8 @@ public interface QueryServices extends SQLCloseable { "phoenix.vector.centroid.cache.maxSize"; // Number of IVF centroids probed per vector query; 0 defaults to sqrt(total lists) public static final String VECTOR_PROBE_COUNT_ATTRIB = "phoenix.vector.probe.count"; + // Maximum number of probe batches scanned during adaptive vector query execution + public static final String VECTOR_MAX_PROBE_LIMIT_ATTRIB = "phoenix.vector.max.probe.limit"; public static final String HA_GROUP_NAME_ATTRIB = "phoenix.ha.group"; public static final String AUTO_UPGRADE_WHITELIST_ATTRIB = "phoenix.client.autoUpgradeWhiteList"; // Mainly for testing to force spilling diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java index 1b62655109..cad482412f 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java @@ -234,6 +234,7 @@ public class QueryServicesOptions { public static final long DEFAULT_MAX_CLIENT_METADATA_CACHE_SIZE = 1024L * 1024L * 10L; // 10 Mb public static final long DEFAULT_VECTOR_CENTROID_CACHE_MAX_SIZE = 1000L; public static final int DEFAULT_VECTOR_PROBE_COUNT = 0; + public static final int DEFAULT_VECTOR_MAX_PROBE_LIMIT = 8; public static final int DEFAULT_GROUPBY_ESTIMATED_DISTINCT_VALUES = 1000; public static final int DEFAULT_CLOCK_SKEW_INTERVAL = 2000; public static final boolean DEFAULT_INDEX_FAILURE_HANDLING_REBUILD = true; // auto rebuild on diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorHybridFilterIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorHybridFilterIT.java new file mode 100644 index 0000000000..6d051534a7 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorHybridFilterIT.java @@ -0,0 +1,416 @@ +/* + * 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.phoenix.end2end; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import org.apache.phoenix.compile.QueryPlan; +import org.apache.phoenix.execute.VectorIndexScanPlan; +import org.apache.phoenix.jdbc.PhoenixPreparedStatement; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.util.QueryUtil; +import org.bson.BinaryVector; +import org.bson.BsonBinary; +import org.bson.BsonDocument; +import org.bson.BsonString; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Integration tests for hybrid vector queries with relational filters, verifying adaptive centroid + * probing, selective secondary index pre-filtering, and covered vs. uncovered document predicate + * evaluation. + */ +@Category(ParallelStatsDisabledTest.class) +public class VectorHybridFilterIT extends ParallelStatsDisabledIT { + + private static final List<float[]> CENTROIDS = + Arrays.asList(new float[] { 0f, 0f, 0f, 0f }, new float[] { 10f, 0f, 0f, 0f }, + new float[] { 20f, 0f, 0f, 0f }, new float[] { 30f, 0f, 0f, 0f }); + private static final float[] ORIGIN = new float[] { 0f, 0f, 0f, 0f }; + + private static Float[] boxed(float[] v) { + Float[] b = new Float[v.length]; + for (int i = 0; i < v.length; i++) { + b[i] = v[i]; + } + return b; + } + + /** + * Test fixture with four centroid clusters partitioned across category values to validate + * selective filtering and deferred projection across posting lists. + */ + private static void createFixture(Connection conn, String table, String index) + throws SQLException { + createFixture(conn, table, index, ""); + } + + private static void createFixture(Connection conn, String table, String index, + String tableOptions) throws SQLException { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + table + " (ID VARCHAR NOT NULL PRIMARY KEY, " + + "V VECTOR(FLOAT, 4), CATEGORY VARCHAR, DESCRIPTION VARCHAR) " + tableOptions); + stmt.execute("CREATE VECTOR INDEX " + index + " ON " + table + " (V) INCLUDE (CATEGORY) " + + "WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)"); + } + VectorIndexTestUtil.activateWithKnownCentroids(conn, table, index, CENTROIDS); + try (PreparedStatement ps = conn.prepareStatement( + "UPSERT INTO " + table + " (ID, V, CATEGORY, DESCRIPTION) VALUES (?, ?, ?, ?)")) { + for (int c = 0; c < 4; c++) { + for (int a = 1; a <= 4; a++) { + upsert(conn, ps, "c" + c + "_a" + a, c * 10f + a * 0.1f, "A"); + } + upsert(conn, ps, "c" + c + "_b1", c * 10f + 0.5f, "B"); + } + } + conn.commit(); + } + + private static void upsert(Connection conn, PreparedStatement ps, String id, float x, + String category) throws SQLException { + ps.setString(1, id); + ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { x, 0f, 0f, 0f })); + ps.setString(3, category); + ps.setString(4, "description of " + id); + ps.executeUpdate(); + } + + private static List<String> search(Connection conn, String sql, float[] q) throws SQLException { + List<String> ids = new ArrayList<>(); + try (PreparedStatement ps = conn.prepareStatement(sql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxed(q))); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + ids.add(rs.getString(1)); + } + } + } + return ids; + } + + private static String explain(Connection conn, String sql, float[] q) throws SQLException { + try (PreparedStatement ps = conn.prepareStatement("EXPLAIN " + sql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxed(q))); + return QueryUtil.getExplainPlan(ps.executeQuery()); + } + } + + private static QueryPlan optimize(Connection conn, String sql, float[] q) throws SQLException { + PreparedStatement ps = conn.prepareStatement(sql); + ps.setArray(1, conn.createArrayOf("FLOAT", boxed(q))); + return ps.unwrap(PhoenixPreparedStatement.class).optimizeQuery(); + } + + @Test + public void testAdaptiveProbingSatisfiesLimit() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index); + // Requires expansion across multiple centroid posting lists to satisfy the limit + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID, CATEGORY FROM " + table + + " WHERE CATEGORY = 'B' ORDER BY L2_DISTANCE(V, ?) LIMIT 3"; + assertEquals(Arrays.asList("c0_b1", "c1_b1", "c2_b1"), search(conn, sql, ORIGIN)); + String plan = explain(conn, sql, ORIGIN); + assertTrue(plan, + plan.contains("CLIENT PROBING 1 OF 4 CENTROIDS (L2) EXPANDING UP TO 4 BATCHES")); + } + } + + @Test + public void testAdaptiveProbingAppliesOffsetToGlobalOrder() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index); + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID FROM " + table + + " WHERE CATEGORY = 'B' ORDER BY L2_DISTANCE(V, ?) LIMIT 2 OFFSET 1"; + assertEquals(Arrays.asList("c1_b1", "c2_b1"), search(conn, sql, ORIGIN)); + } + } + + @Test + public void testAdaptiveProbingStopsAtEveryCentroid() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index); + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID FROM " + table + + " WHERE CATEGORY = 'B' ORDER BY L2_DISTANCE(V, ?) LIMIT 10"; + assertEquals(Arrays.asList("c0_b1", "c1_b1", "c2_b1", "c3_b1"), search(conn, sql, ORIGIN)); + } + } + + @Test + public void testAdaptiveProbingStopsAtMaxProbeLimit() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index); + String hinted = "SELECT /*+ VECTOR_PROBE_COUNT(1) MAX_PROBE_LIMIT(2) */ ID FROM " + table + + " WHERE CATEGORY = 'B' ORDER BY L2_DISTANCE(V, ?) LIMIT 3"; + assertEquals(Arrays.asList("c0_b1", "c1_b1"), search(conn, hinted, ORIGIN)); + assertTrue(explain(conn, hinted, ORIGIN).contains("EXPANDING UP TO 2 BATCHES")); + + Properties props = new Properties(); + props.setProperty(QueryServices.VECTOR_MAX_PROBE_LIMIT_ATTRIB, "2"); + try (Connection limited = DriverManager.getConnection(getUrl(), props)) { + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID FROM " + table + + " WHERE CATEGORY = 'B' ORDER BY L2_DISTANCE(V, ?) LIMIT 3"; + assertEquals(Arrays.asList("c0_b1", "c1_b1"), search(limited, sql, ORIGIN)); + } + // Single-batch limit disables adaptive expansion + String single = "SELECT /*+ VECTOR_PROBE_COUNT(1) MAX_PROBE_LIMIT(1) */ ID FROM " + table + + " WHERE CATEGORY = 'B' ORDER BY L2_DISTANCE(V, ?) LIMIT 3"; + assertEquals(Arrays.asList("c0_b1"), search(conn, single, ORIGIN)); + assertFalse(explain(conn, single, ORIGIN).contains("EXPANDING")); + } + } + + @Test + public void testUnfilteredQueryDoesNotProbeAdaptively() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index); + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID FROM " + table + + " ORDER BY L2_DISTANCE(V, ?) LIMIT 10"; + // Unfiltered queries execute without adaptive expansion + assertEquals(Arrays.asList("c0_a1", "c0_a2", "c0_a3", "c0_a4", "c0_b1"), + search(conn, sql, ORIGIN)); + assertFalse(explain(conn, sql, ORIGIN).contains("EXPANDING")); + } + } + + @Test + public void testAdaptiveProbingWithDeferredProjection() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index); + // Covered predicate evaluated during index probing with deferred projection for uncovered + // attributes + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID, DESCRIPTION FROM " + table + + " WHERE CATEGORY = 'B' ORDER BY L2_DISTANCE(V, ?) LIMIT 3"; + QueryPlan plan = optimize(conn, sql, ORIGIN); + assertTrue(plan.getClass().getName(), VectorIndexTestUtil.isDeferredProjection(plan)); + Map<String, String> rows = new LinkedHashMap<>(); + try (PreparedStatement ps = conn.prepareStatement(sql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxed(ORIGIN))); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + rows.put(rs.getString(1), rs.getString(2)); + } + } + } + assertEquals(Arrays.asList("c0_b1", "c1_b1", "c2_b1"), new ArrayList<>(rows.keySet())); + for (Map.Entry<String, String> row : rows.entrySet()) { + assertEquals("description of " + row.getKey(), row.getValue()); + } + } + } + + @Test + public void testFilteredPlanHasEstimates() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index, "GUIDE_POSTS_WIDTH=20"); + try (Statement stmt = conn.createStatement()) { + stmt.execute("UPDATE STATISTICS " + table); + } + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID FROM " + table + + " WHERE CATEGORY = 'B' ORDER BY L2_DISTANCE(V, ?) LIMIT 3"; + VectorIndexScanPlan plan = VectorIndexTestUtil.vectorPlan(optimize(conn, sql, ORIGIN)); + assertTrue(plan.isAdaptive()); + assertNotNull("adaptive plan must estimate its first batch", plan.getEstimatedRowsToScan()); + assertFalse(plan.getCost().isUnknown()); + } + } + + /** + * Test fixture evaluating optimizer plan selection between secondary index filtering and vector + * index probing across varying selectivity distributions. + */ + private static void createAuthorFixture(Connection conn, String table, String authorIndex, + String vectorIndex, boolean selective, boolean coverVector) throws SQLException { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + table + " (ID VARCHAR NOT NULL PRIMARY KEY, " + + "V VECTOR(FLOAT, 4), AUTHOR VARCHAR) GUIDE_POSTS_WIDTH=100"); + stmt.execute("CREATE INDEX " + authorIndex + " ON " + table + " (AUTHOR)" + + (coverVector ? " INCLUDE (V)" : "")); + stmt.execute("CREATE VECTOR INDEX " + vectorIndex + " ON " + table + " (V) INCLUDE (AUTHOR)" + + " WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)"); + } + VectorIndexTestUtil.activateWithKnownCentroids(conn, table, vectorIndex, CENTROIDS); + try (PreparedStatement ps = + conn.prepareStatement("UPSERT INTO " + table + " (ID, V, AUTHOR) VALUES (?, ?, ?)")) { + for (int i = 0; i < 1000; i++) { + ps.setString(1, String.format("row_%04d", i)); + ps.setArray(2, + conn.createArrayOf("FLOAT", new Float[] { (i % 4) * 10f + i * 0.001f, 0f, 0f, 0f })); + ps.setString(3, i < (selective ? 10 : 900) ? "Alice" : "Bob"); + ps.executeUpdate(); + if (i % 250 == 0) { + conn.commit(); + } + } + } + conn.commit(); + try (Statement stmt = conn.createStatement()) { + stmt.execute("UPDATE STATISTICS " + table); + } + } + + private static List<String> bruteForceAlice(Connection conn, String table) throws SQLException { + return search(conn, "SELECT /*+ NO_INDEX */ ID FROM " + table + + " WHERE AUTHOR = 'Alice' ORDER BY L2_DISTANCE(V, ?) LIMIT 5", ORIGIN); + } + + @Test + public void testSelectiveFilterIsEvaluatedFirst() throws Exception { + String table = generateUniqueName(); + String authorIndex = generateUniqueName(); + String vectorIndex = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createAuthorFixture(conn, table, authorIndex, vectorIndex, true, true); + String sql = + "SELECT ID FROM " + table + " WHERE AUTHOR = 'Alice' ORDER BY L2_DISTANCE(V, ?) LIMIT 5"; + String plan = explain(conn, sql, ORIGIN); + assertTrue(plan, plan.contains(authorIndex)); + assertFalse(plan, plan.contains("CLIENT PROBING")); + // Pre-filtered secondary index scan produces exact results matching brute force scan + assertEquals(bruteForceAlice(conn, table), search(conn, sql, ORIGIN)); + } + } + + @Test + public void testUncoveredSelectiveIndexIsUsedWhenHinted() throws Exception { + String table = generateUniqueName(); + String authorIndex = generateUniqueName(); + String vectorIndex = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createAuthorFixture(conn, table, authorIndex, vectorIndex, true, false); + // Uncovered secondary index requires explicit hint to be selected + String sql = + "SELECT ID FROM " + table + " WHERE AUTHOR = 'Alice' ORDER BY L2_DISTANCE(V, ?) LIMIT 5"; + assertTrue(explain(conn, sql, ORIGIN).contains(vectorIndex)); + String hinted = "SELECT /*+ INDEX(" + table + " " + authorIndex + ") */ ID FROM " + table + + " WHERE AUTHOR = 'Alice' ORDER BY L2_DISTANCE(V, ?) LIMIT 5"; + String plan = explain(conn, hinted, ORIGIN); + assertTrue(plan, plan.contains(authorIndex)); + assertFalse(plan, plan.contains("CLIENT PROBING")); + assertEquals(bruteForceAlice(conn, table), search(conn, hinted, ORIGIN)); + } + } + + @Test + public void testUnselectiveFilterUsesVectorIndex() throws Exception { + String table = generateUniqueName(); + String authorIndex = generateUniqueName(); + String vectorIndex = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createAuthorFixture(conn, table, authorIndex, vectorIndex, false, true); + String sql = + "SELECT ID FROM " + table + " WHERE AUTHOR = 'Alice' ORDER BY L2_DISTANCE(V, ?) LIMIT 5"; + String plan = explain(conn, sql, ORIGIN); + assertTrue(plan, plan.contains(vectorIndex)); + assertTrue(plan, plan.contains("CLIENT PROBING")); + assertFalse(plan, plan.contains(authorIndex)); + } + } + + private static BsonDocument doc(float x, int i, String category) { + float[] v = new float[] { x, 0.1f * i, 0f, 0f }; + BsonDocument doc = new BsonDocument("embedding", new BsonBinary(BinaryVector.floatVector(v))); + doc.put("category", new BsonString(category)); + return doc; + } + + private static void testDocumentFilter(boolean coverDocument) throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + table + + " (PK VARCHAR NOT NULL PRIMARY KEY, DOC BSON) IMMUTABLE_ROWS=true"); + stmt.execute("CREATE VECTOR INDEX " + index + " ON " + table + + " (BSON_VECTOR_VALUE(DOC, 'embedding', 4))" + (coverDocument ? " INCLUDE (DOC)" : "") + + " WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)"); + } + VectorIndexTestUtil.activateWithKnownCentroids(conn, table, index, CENTROIDS); + float[] xs = { 1f, 3f, 4f, 9f, 11f, 13f, 19f, 21f, 23f, 28f, 29f, 31f }; + String[] categories = { "science", "art", "science", "art", "math", "science", "art", "math", + "science", "art", "math", "science" }; + try ( + PreparedStatement ps = conn.prepareStatement("UPSERT INTO " + table + " VALUES (?, ?)")) { + for (int i = 0; i < xs.length; i++) { + ps.setString(1, "r" + i); + ps.setObject(2, doc(xs[i], i, categories[i])); + ps.executeUpdate(); + } + } + conn.commit(); + + float[] q = new float[] { 5f, 0f, 0f, 0f }; + String vector = "BSON_VECTOR_VALUE(DOC, 'embedding', 4)"; + for (String predicate : new String[] { "BSON_VALUE(DOC, 'category', 'VARCHAR') = 'science'", + "BSON_VALUE(DOC, 'category', 'VARCHAR') IN ('science', 'math')" }) { + String sql = "SELECT PK FROM " + table + " WHERE " + predicate + " ORDER BY L2_DISTANCE(" + + vector + ", ?) LIMIT 4"; + String plan = explain(conn, sql, q); + assertTrue(plan, plan.contains(index)); + assertTrue(plan, plan.contains("CLIENT PROBING")); + // Server merge required when document column is not covered by the vector index + assertEquals(plan, !coverDocument, plan.contains("SERVER MERGE")); + String exact = "SELECT /*+ NO_INDEX */ PK FROM " + table + " WHERE " + predicate + + " ORDER BY L2_DISTANCE(" + vector + ", ?) LIMIT 4"; + String all = "SELECT /*+ VECTOR_PROBE_COUNT(4) */ PK FROM " + table + " WHERE " + predicate + + " ORDER BY L2_DISTANCE(" + vector + ", ?) LIMIT 4"; + assertEquals(search(conn, exact, q), search(conn, all, q)); + assertEquals(search(conn, exact, q), search(conn, sql, q)); + } + } + } + + @Test + public void testCoveredDocumentFilter() throws Exception { + testDocumentFilter(true); + } + + @Test + public void testUncoveredDocumentFilter() throws Exception { + testDocumentFilter(false); + } +} diff --git a/phoenix-core/src/test/java/org/apache/phoenix/execute/VectorIndexScanPlanTest.java b/phoenix-core/src/test/java/org/apache/phoenix/execute/VectorIndexScanPlanTest.java index 9a8d9de9f6..d3b81df4f0 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/execute/VectorIndexScanPlanTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/execute/VectorIndexScanPlanTest.java @@ -199,8 +199,9 @@ public class VectorIndexScanPlanTest extends BaseConnectionlessQueryTest { @Test public void testProbeHints() { - HintNode hint = new HintNode("/*+ VECTOR_PROBE_COUNT(5) */"); + HintNode hint = new HintNode("/*+ VECTOR_PROBE_COUNT(5) MAX_PROBE_LIMIT(3) */"); assertEquals(5, VectorIndexScanPlan.getHintInt(hint, Hint.VECTOR_PROBE_COUNT, 0)); + assertEquals(3, VectorIndexScanPlan.getHintInt(hint, Hint.MAX_PROBE_LIMIT, 8)); HintNode none = new HintNode("/*+ INDEX(T I) */"); assertEquals(7, VectorIndexScanPlan.getHintInt(none, Hint.VECTOR_PROBE_COUNT, 7)); HintNode invalid = new HintNode("/*+ VECTOR_PROBE_COUNT(abc) */"); diff --git a/phoenix-core/src/test/java/org/apache/phoenix/optimize/VectorPlanOrderingTest.java b/phoenix-core/src/test/java/org/apache/phoenix/optimize/VectorPlanOrderingTest.java index 5a6ae17702..3b79e3b033 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/optimize/VectorPlanOrderingTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/optimize/VectorPlanOrderingTest.java @@ -24,12 +24,17 @@ import static org.mockito.Mockito.when; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; +import java.util.Set; import org.apache.phoenix.compile.QueryPlan; import org.apache.phoenix.compile.StatementContext; import org.apache.phoenix.execute.HashJoinPlan; import org.apache.phoenix.execute.HashJoinPlan.SubPlan; import org.apache.phoenix.execute.VectorIndexScanPlan; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTableType; +import org.apache.phoenix.schema.TableRef; import org.junit.Test; /** Unit tests for vector index plan comparison and precedence rules in query optimization. */ @@ -98,4 +103,44 @@ public class VectorPlanOrderingTest { assertTrue(QueryOptimizer.compareVectorPlans(data, vector, DATA_HINTED) < 0); assertEquals(0, QueryOptimizer.compareVectorPlans(index, data, DATA_HINTED)); } + + private static QueryPlan planOver(PTableType type, Long estimatedRows) throws Exception { + QueryPlan plan = mock(QueryPlan.class); + PTable table = mock(PTable.class); + when(table.getType()).thenReturn(type); + TableRef tableRef = mock(TableRef.class); + when(tableRef.getTable()).thenReturn(table); + when(plan.getTableRef()).thenReturn(tableRef); + when(plan.getEstimatedRowsToScan()).thenReturn(estimatedRows); + return plan; + } + + @Test + public void testSelectiveRegularIndexIsEvaluatedFirst() throws Exception { + QueryPlan data = planOver(PTableType.TABLE, 10_000L); + QueryPlan selective = planOver(PTableType.INDEX, 10L); + QueryPlan unselective = planOver(PTableType.INDEX, 9_000L); + QueryPlan vector = vectorScan(false); + List<QueryPlan> plans = Arrays.asList(data, selective, unselective, vector); + Set<QueryPlan> filterFirst = VectorSearchUtil.getFilterFirstPlans(plans, data); + assertEquals(Collections.singleton(selective), filterFirst); + + List<QueryPlan> ordered = new ArrayList<>(Arrays.asList(unselective, vector, data, selective)); + ordered.sort((p1, p2) -> QueryOptimizer.compareVectorPlans(p1, p2, NO_HINT, filterFirst)); + assertEquals(selective, ordered.get(0)); + assertEquals(vector, ordered.get(1)); + // Explicit data table hint overrides secondary index filter first precedence + assertEquals(0, QueryOptimizer.compareVectorPlans(selective, data, DATA_HINTED, filterFirst)); + assertTrue(QueryOptimizer.compareVectorPlans(selective, vector, DATA_HINTED, filterFirst) < 0); + } + + @Test + public void testNoFilterFirstWithoutStatisticsOrVectorPlan() throws Exception { + QueryPlan noStats = planOver(PTableType.TABLE, null); + QueryPlan index = planOver(PTableType.INDEX, 1L); + assertTrue(VectorSearchUtil + .getFilterFirstPlans(Arrays.asList(noStats, index, vectorScan(false)), noStats).isEmpty()); + QueryPlan data = planOver(PTableType.TABLE, 10_000L); + assertTrue(VectorSearchUtil.getFilterFirstPlans(Arrays.asList(data, index), data).isEmpty()); + } }
