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 da30e97fff849f64db7d959c825e1603c47e04d4 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 | 21 +- .../phoenix/execute/VectorIndexScanPlan.java | 277 +++++++++++- .../apache/phoenix/optimize/OptimizerReasons.java | 1 + .../apache/phoenix/optimize/QueryOptimizer.java | 68 ++- .../apache/phoenix/optimize/VectorSearchUtil.java | 122 +++++- .../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 | 465 +++++++++++++++++++++ .../phoenix/execute/VectorIndexScanPlanTest.java | 3 +- .../phoenix/optimize/VectorPlanOrderingTest.java | 129 ++++++ 11 files changed, 1059 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..d3912c8c45 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,19 @@ public class ScanPlan extends BaseQueryPlan { && ScanUtil.shouldRowsBeInRowKeyOrder(orderBy, context)); } + /** + * Records the row and byte estimates, the splits and the scans of the given result iterators in + * this plan. A subclass that makes its own iterators calls this method, so that the plan can + * report its estimates. + */ + 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 +323,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 578957f87c..20e5008c2c 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,10 +40,20 @@ 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.index.vector.CentroidManager; +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; @@ -43,12 +61,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; @@ -75,6 +96,12 @@ 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; + // Skip scan filter for the posting lists of the first probe batch, or null. Each later adaptive + // batch replaces this filter in its copy of the scan. + private final SkipScanFilter postingListFilter; public VectorIndexScanPlan(StatementContext context, FilterableStatement statement, TableRef table, RowProjector projector, Integer limit, Integer offset, OrderBy orderBy, @@ -87,7 +114,7 @@ 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()) @@ -95,8 +122,10 @@ public class VectorIndexScanPlan extends ScanPlan { 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)); @@ -109,12 +138,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)); } /** @@ -166,7 +200,19 @@ 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); + } + + /** + * Makes scan ranges for the posting lists of the given centroids. A salted index gets a slot with + * all salt buckets. A multitenant index gets a slot with the tenant ID of the connection, so the + * caller must make sure that the connection has a tenant ID. + */ + 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())); } @@ -250,8 +296,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 @@ -260,7 +311,221 @@ public class VectorIndexScanPlan extends ScanPlan { .setVectorCentroidCount(centroidCount).setVectorDistanceMetric(metric).build()); } - /** Number of probed centroid posting lists, or 0 for full index scans. */ + /** + * Returns true if execution can scan more centroid posting lists when the probed lists hold fewer + * rows than the limit. A filter can cause this. A tenant scope can also cause this, because all + * tenants share the centroids. The plan must have a LIMIT, must probe fewer centroids than the + * index has, and must allow more than one probe batch. + */ + public boolean isAdaptive() { + return limit != null && probeCount > 0 && probeCount < centroidCount && 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); + // Make the first probe batch now, so that the plan records its estimates 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); + } + + /** + * Returns the scan filter with {@code next} in place of the posting list filter of the first + * batch. A null {@code next} removes that filter. The other filters keep their structure and + * position. The method finds the posting list filter only at the top level or in a top level + * FilterList. If it does not find the filter, it puts {@code next} first with MUST_PASS_ALL. + */ + 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); + // The iterators of this batch read their scan ranges from the statement context + 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); + } + } + + /** Orders tuples on the client by the compiled ORDER BY expressions and their null order. */ + 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; + } + } + + /** Returns the maximum number of probe batches, including the first, that a query can scan. */ + public int getMaxProbeBatches() { + return maxProbeBatches; + } + + /** + * Returns the number of posting lists that the first probe batch scans, or 0 if the plan 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 2b44d05e54..3a69e51f10 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 b54584bc10..d20e3bf092 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 @@ -932,13 +932,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)); // If the costs are equal, the static precedence rules for vector plans set the order. - 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 @@ -1010,7 +1013,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(); @@ -1080,30 +1083,45 @@ 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. + * Orders candidate plans for a nearest neighbor vector search by access tier and index coverage. + * Without a data table hint, filter first plans ({@link VectorSearchUtil#getFilterFirstPlans}) + * rank first, because they read few rows and give exact results. Vector index plans rank next, + * before all other plans. A data table hint removes both precedences: filter first plans rank + * with the other plans, and vector index plans rank last. Among vector index plans, a plan with + * fewer data table lookups ranks first: a covering index, then deferred projection, then an + * uncovered index scan. Two plans in the same tier that do not read a vector index compare as 0. + * The caller then orders them by other criteria, and the comparison stays transitive. */ 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; } /** @@ -1291,12 +1309,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 62405217c2..b0130552dd 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,10 +18,16 @@ 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; +import org.apache.phoenix.compile.ScanRanges; +import org.apache.phoenix.compile.StatementContext; import org.apache.phoenix.execute.DelegateQueryPlan; import org.apache.phoenix.execute.HashJoinPlan; import org.apache.phoenix.execute.HashJoinPlan.SubPlan; @@ -38,9 +44,16 @@ 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.query.KeyRange; +import org.apache.phoenix.schema.PColumn; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTableType; +import org.apache.phoenix.schema.stats.GuidePostsInfo; +import org.apache.phoenix.schema.stats.GuidePostsKey; import org.apache.phoenix.schema.types.PDataType; import org.apache.phoenix.schema.types.PVectorDataType; import org.apache.phoenix.util.CostUtil; +import org.apache.phoenix.util.SchemaUtil; /** * Finds vector similarity search queries and extracts their parameters. @@ -260,7 +273,114 @@ public final class VectorSearchUtil { : null; } - /** Indicates whether the query execution plan incorporates a vector index scan. */ + /** + * The maximum fraction of rows that a filter first plan can scan. A secondary index plan uses the + * row estimate of the data plan as the base. The data plan uses the row count of the table. + */ + public static final double FILTER_FIRST_MAX_ROW_FRACTION = 0.05; + + /** + * Finds the filter first plans, which read few rows and give exact results, so the optimizer + * prefers them to an approximate vector index plan. A secondary index plan qualifies if + * statistics estimate its rows at no more than {@link #FILTER_FIRST_MAX_ROW_FRACTION} of the data + * plan rows. The data plan qualifies if it binds a primary key column after the constant key + * prefix, and the bound range is small. Returns an empty set if no plan reads a vector index. The + * returned set compares plans by identity. + */ + 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) { + // Without a data plan row estimate, no secondary index plan can be a filter first plan + } + if (isSelectiveKeyRange(dataPlan, dataRows)) { + filterFirst.add(dataPlan); + } + 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) { + // A plan without a row estimate cannot be a filter first plan + } + } + return filterFirst; + } + + /** + * Returns true if the data plan binds a small range of a primary key column. The column must come + * after the salt byte, the tenant ID and the view constants. With statistics, the range must hold + * no more than {@link #FILTER_FIRST_MAX_ROW_FRACTION} of the table rows. Without statistics, the + * first such column must be bound to point keys. A vector plan scans the posting lists of all + * keys and applies the bound columns only after it ranks the rows. + */ + private static boolean isSelectiveKeyRange(QueryPlan dataPlan, Long dataRows) { + StatementContext context = dataPlan.getContext(); + ScanRanges ranges = context.getScanRanges(); + PTable table = dataPlan.getTableRef().getTable(); + int prefix = (ranges.isSalted() ? 1 : 0) + + (table.isMultiTenant() && context.getConnection().getTenantId() != null ? 1 : 0); + for (PColumn column : table.getPKColumns()) { + prefix += column.getViewConstant() == null ? 0 : 1; + } + if (ranges.getBoundPkColumnCount() <= prefix) { + return false; + } + Long tableRows = null; + try { + GuidePostsInfo stats = + context.getConnection().getQueryServices().getTableStats(new GuidePostsKey( + table.getPhysicalName().getBytes(), SchemaUtil.getEmptyColumnFamily(table))); + long rows = 0; + for (long count : stats.getRowCounts()) { + rows += count; + } + tableRows = rows > 0 ? rows : null; + } catch (SQLException e) { + // Without table statistics, only point keys can qualify the bound key range + } + if (dataRows != null && tableRows != null) { + return dataRows <= tableRows * FILTER_FIRST_MAX_ROW_FRACTION; + } + // Without statistics, only point keys qualify. A range such as ID > ? on the first column + // after the prefix can span most of the table. + int[] spans = ranges.getSlotSpans(); + List<List<KeyRange>> slots = ranges.getRanges(); + for (int i = 0, column = 0; i < slots.size(); i++) { + column += spans[i] + 1; + if (column > prefix) { + for (KeyRange range : slots.get(i)) { + if (!range.isSingleKey()) { + return false; + } + } + return true; + } + } + return false; + } + + /** Returns true if 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 649f22c199..c953f995cf 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 { /** * Sets the number of centroids that a vector index scan probes in each model generation. */ - VECTOR_PROBE_COUNT; + VECTOR_PROBE_COUNT, + + /** + * Overrides the maximum number of probe batches that an adaptive vector index search can scan. + */ + 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 ad40622312..f932aea5a8 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 @@ -155,6 +155,8 @@ public interface QueryServices extends SQLCloseable { "phoenix.vector.centroid.cache.maxBytes"; // Centroids that a vector query probes in each generation; 0 selects sqrt(centroid count) public static final String VECTOR_PROBE_COUNT_ATTRIB = "phoenix.vector.probe.count"; + // Maximum number of probe batches, including the first, that an adaptive vector query can scan + 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 23a37298f7..c035ca8b24 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_BYTES = 1024L * 1024L * 256L; // 256 Mb 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..d45e98fd08 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorHybridFilterIT.java @@ -0,0 +1,465 @@ +/* + * 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.PhoenixRuntime; +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 vector queries that also have relational filters. The tests examine + * adaptive centroid probing, filter first plans on selective secondary indexes, and document + * predicates that the vector index covers or does not cover. + */ +@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; + } + + /** + * Creates a table with four centroid clusters. Each cluster has four rows with CATEGORY 'A' and + * one row with CATEGORY 'B'. The vector index covers CATEGORY but not DESCRIPTION, so the tests + * can examine filtered probes and deferred projection. + */ + 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); + // Each posting list holds one 'B' row, so the scan must expand 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)); + } + // A limit of one probe batch 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 testUnfilteredQueryProbesAdaptively() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index); + // The nearest posting list holds five rows, so a limit of ten expands to the next list + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID FROM " + table + + " ORDER BY L2_DISTANCE(V, ?) LIMIT 10"; + assertEquals(Arrays.asList("c0_a1", "c0_a2", "c0_a3", "c0_a4", "c0_b1", "c1_a1", "c1_a2", + "c1_a3", "c1_a4", "c1_b1"), search(conn, sql, ORIGIN)); + assertTrue(explain(conn, sql, ORIGIN).contains("EXPANDING UP TO 4 BATCHES")); + } + } + + @Test + public void testTenantScopedQueryProbesAdaptively() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + table + " (TENANT_ID VARCHAR NOT NULL, " + + "ID VARCHAR NOT NULL, V VECTOR(FLOAT, 4) CONSTRAINT PK PRIMARY KEY (TENANT_ID, ID)) " + + "MULTI_TENANT=true"); + stmt.execute("CREATE VECTOR INDEX " + index + " ON " + table + + " (V) WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)"); + } + VectorIndexTestUtil.activateWithKnownCentroids(conn, table, index, CENTROIDS); + // A large tenant fills all posting lists. A small tenant has one row in each list. + try (PreparedStatement ps = + conn.prepareStatement("UPSERT INTO " + table + " (TENANT_ID, ID, V) VALUES (?, ?, ?)")) { + for (int c = 0; c < 4; c++) { + for (int a = 1; a <= 5; a++) { + ps.setString(1, "LARGE"); + ps.setString(2, "c" + c + "_" + a); + ps.setArray(3, + conn.createArrayOf("FLOAT", new Float[] { c * 10f + a * 0.1f, 0f, 0f, 0f })); + ps.executeUpdate(); + } + ps.setString(1, "SMALL"); + ps.setString(2, "c" + c); + ps.setArray(3, conn.createArrayOf("FLOAT", new Float[] { c * 10f + 0.3f, 0f, 0f, 0f })); + ps.executeUpdate(); + } + } + conn.commit(); + } + Properties props = new Properties(); + props.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, "SMALL"); + try (Connection tenant = DriverManager.getConnection(getUrl(), props)) { + // All tenants share the centroids. The nearest posting list holds only one row of this + // tenant, so the scan must expand to satisfy the limit. + String sql = "SELECT /*+ VECTOR_PROBE_COUNT(1) */ ID FROM " + table + + " ORDER BY L2_DISTANCE(V, ?) LIMIT 3"; + VectorIndexScanPlan plan = VectorIndexTestUtil.vectorPlan(optimize(tenant, sql, ORIGIN)); + assertNotNull(plan); + assertTrue(plan.isAdaptive()); + assertEquals(Arrays.asList("c0", "c1", "c2"), search(tenant, sql, ORIGIN)); + } + } + + @Test + public void testAdaptiveProbingWithDeferredProjection() throws Exception { + String table = generateUniqueName(); + String index = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createFixture(conn, table, index); + // The index scan evaluates the covered CATEGORY predicate. Deferred projection reads the + // uncovered DESCRIPTION column from the data table. + 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()); + } + } + + /** + * Creates a table with a secondary index on AUTHOR and a vector index, and collects statistics. + * If {@code selective} is true, 10 of 1000 rows have AUTHOR 'Alice', otherwise 900 rows do. If + * {@code coverVector} is true, the AUTHOR index covers the vector column. + */ + 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")); + // The secondary index scan filters first, so its results are exact and match a full 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); + // The optimizer selects an uncovered secondary index only if a hint names it + 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")); + // A server merge is necessary if the vector index does not cover the document column + 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 a6a3c892c6..21c1c0b00e 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 @@ -203,8 +203,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 e3897a35b8..146219d85e 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 @@ -19,17 +19,31 @@ package org.apache.phoenix.optimize; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; 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.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.hbase.util.Bytes; import org.apache.phoenix.compile.QueryPlan; +import org.apache.phoenix.compile.ScanRanges; 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.jdbc.PhoenixConnection; +import org.apache.phoenix.query.ConnectionQueryServices; +import org.apache.phoenix.query.KeyRange; +import org.apache.phoenix.schema.PNameFactory; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTableType; +import org.apache.phoenix.schema.TableRef; +import org.apache.phoenix.schema.stats.GuidePostsInfo; import org.junit.Test; /** Unit tests for the precedence rules that order vector index plans in the query optimizer. */ @@ -98,4 +112,119 @@ 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 { + return planOver(type, estimatedRows, Collections.emptyList(), null); + } + + private static final KeyRange POINT = KeyRange.getKeyRange(Bytes.toBytes("u1")); + + /** + * Returns a mock plan whose key ranges bind the leading primary key columns, one column for each + * slot of {@code bound}. The table statistics count {@code tableRows} rows. If {@code tableRows} + * is null, the table has no statistics. + */ + private static QueryPlan planOver(PTableType type, Long estimatedRows, List<List<KeyRange>> bound, + Long tableRows) throws Exception { + QueryPlan plan = mock(QueryPlan.class); + PTable table = mock(PTable.class); + when(table.getType()).thenReturn(type); + when(table.getPhysicalName()).thenReturn(PNameFactory.newName("T")); + TableRef tableRef = mock(TableRef.class); + when(tableRef.getTable()).thenReturn(table); + when(plan.getTableRef()).thenReturn(tableRef); + when(plan.getEstimatedRowsToScan()).thenReturn(estimatedRows); + ScanRanges ranges = mock(ScanRanges.class); + when(ranges.getBoundPkColumnCount()).thenReturn(bound.size()); + when(ranges.getRanges()).thenReturn(bound); + when(ranges.getSlotSpans()).thenReturn(new int[bound.size()]); + ConnectionQueryServices services = mock(ConnectionQueryServices.class); + when(services.getTableStats(any())).thenReturn(tableRows == null + ? GuidePostsInfo.NO_GUIDEPOST + : new GuidePostsInfo(Arrays.asList(1L, 1L), new ImmutableBytesWritable(new byte[0]), + Arrays.asList(tableRows / 2, tableRows - tableRows / 2), 0, 2, Arrays.asList(0L, 0L))); + PhoenixConnection connection = mock(PhoenixConnection.class); + when(connection.getQueryServices()).thenReturn(services); + StatementContext context = mock(StatementContext.class); + when(context.getScanRanges()).thenReturn(ranges); + when(context.getConnection()).thenReturn(connection); + when(plan.getContext()).thenReturn(context); + 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)); + // An explicit data table hint removes the filter first precedence of the secondary index + 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()); + } + + @Test + public void testDataPlanBindingPrimaryKeyIsEvaluatedFirst() throws Exception { + // WHERE USER_ID = ? on PK (USER_ID, DOC_ID): the exact range scan ranks before the vector + // plan. The vector plan probes the posting lists of all users and filters USER_ID after that. + QueryPlan vector = vectorScan(false); + QueryPlan bound = planOver(PTableType.TABLE, null, Arrays.asList(Arrays.asList(POINT)), null); + Set<QueryPlan> filterFirst = + VectorSearchUtil.getFilterFirstPlans(Arrays.asList(bound, vector), bound); + assertEquals(Collections.singleton(bound), filterFirst); + assertTrue(QueryOptimizer.compareVectorPlans(bound, vector, NO_HINT, filterFirst) < 0); + assertTrue(QueryOptimizer.compareVectorPlans(bound, vector, DATA_HINTED, filterFirst) < 0); + + // With statistics, the bound range must hold no more than the filter first fraction + QueryPlan narrow = planOver(PTableType.TABLE, 40L, Arrays.asList(Arrays.asList(POINT)), 1_000L); + assertEquals(Collections.singleton(narrow), + VectorSearchUtil.getFilterFirstPlans(Arrays.asList(narrow, vector), narrow)); + QueryPlan wide = planOver(PTableType.TABLE, 600L, Arrays.asList(Arrays.asList(POINT)), 1_000L); + assertTrue(VectorSearchUtil.getFilterFirstPlans(Arrays.asList(wide, vector), wide).isEmpty()); + + // Without statistics, an IN list qualifies because it binds point keys. A range such as + // USER_ID > ? does not qualify, because it can span most of the table. With statistics, the + // same range qualifies if it is narrow. + QueryPlan inList = planOver(PTableType.TABLE, null, + Arrays.asList(Arrays.asList(POINT, KeyRange.getKeyRange(Bytes.toBytes("u2")))), null); + assertEquals(Collections.singleton(inList), + VectorSearchUtil.getFilterFirstPlans(Arrays.asList(inList, vector), inList)); + List<List<KeyRange>> range = Arrays.asList( + Arrays.asList(KeyRange.getKeyRange(Bytes.toBytes("u1"), false, KeyRange.UNBOUND, false))); + QueryPlan unestimated = planOver(PTableType.TABLE, null, range, null); + assertTrue(VectorSearchUtil.getFilterFirstPlans(Arrays.asList(unestimated, vector), unestimated) + .isEmpty()); + QueryPlan narrowRange = planOver(PTableType.TABLE, 40L, range, 1_000L); + assertEquals(Collections.singleton(narrowRange), + VectorSearchUtil.getFilterFirstPlans(Arrays.asList(narrowRange, vector), narrowRange)); + // Equality on the first column after the prefix qualifies, whatever binds the next column + QueryPlan pointThenRange = + planOver(PTableType.TABLE, null, Arrays.asList(Arrays.asList(POINT), range.get(0)), null); + assertEquals(Collections.singleton(pointThenRange), + VectorSearchUtil.getFilterFirstPlans(Arrays.asList(pointThenRange, vector), pointThenRange)); + + // A salt byte that is bound to all buckets does not make a selective key range + QueryPlan salted = planOver(PTableType.TABLE, null, Arrays.asList(Arrays.asList(POINT)), null); + when(salted.getContext().getScanRanges().isSalted()).thenReturn(true); + assertTrue( + VectorSearchUtil.getFilterFirstPlans(Arrays.asList(salted, vector), salted).isEmpty()); + } }
