apurtell commented on code in PR #2624: URL: https://github.com/apache/phoenix/pull/2624#discussion_r4198321004
########## phoenix-core-client/src/main/java/org/apache/phoenix/execute/VectorIndexScanPlan.java: ########## @@ -0,0 +1,607 @@ +/* + * 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.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.hadoop.hbase.util.Bytes; +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; +import org.apache.phoenix.compile.ExplainPlanAttributes; +import org.apache.phoenix.compile.ExplainPlanAttributes.ExplainPlanAttributesBuilder; +import org.apache.phoenix.compile.OrderByCompiler.OrderBy; +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; +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; +import org.apache.phoenix.util.SchemaUtil; + +import org.apache.phoenix.thirdparty.com.google.common.base.Optional; + +/** + * Scan plan implementation for IVF vector index queries. Maps query vectors to nearest centroid + * partitions and scopes the scan to corresponding posting list key ranges across salting and tenant + * prefixes. Region servers score candidates within targeted partitions and return ranked streams + * for client side top-K merge sort. + * <p> + * Falls back to full index scans when partition pruning cannot be applied (such as untrained + * indexes, unscoped multitenant queries, or conflicting primary key filter constraints). + */ +public class VectorIndexScanPlan extends ScanPlan { + + /** Defines query compensation policy during concurrent index rebuild migrations. */ + public enum RebuildProbePolicy { + /** Expands candidate probe count by a configured multiplier. */ + EXPAND, + /** Performs exact evaluation across all index rows. */ + EXACT, + /** Standard probe count without rebuild compensation. */ + NONE + } + + private final int centroidCount; + private final int probeCount; + // Base probe count prior to applying rebuild compensation + private final int baseProbeCount; + // Rebuild compensation policy applied during query execution, or null if inactive + private final RebuildProbePolicy rebuildPolicy; + private final int[] probeCentroids; + private final double oversampleFactor; + 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, + ParallelIteratorFactory parallelIteratorFactory, boolean allowPageFilter, QueryPlan dataPlan, + Optional<byte[]> rowOffset) throws SQLException { + super(context, statement, table, projector, limit, offset, orderBy, parallelIteratorFactory, + allowPageFilter, dataPlan, rowOffset); + PTable index = table.getTable(); + PhoenixConnection connection = context.getConnection(); + ReadOnlyProps props = connection.getQueryServices().getProps(); + HintNode hint = statement.getHint(); + + 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; + // Compensate for unindexed rows during active rebuild migrations + this.rebuildPolicy = index.isVectorRebuildInProgress() + ? RebuildProbePolicy.valueOf(props.get(QueryServices.VECTOR_REBUILD_PROBE_POLICY_ATTRIB, + QueryServicesOptions.DEFAULT_VECTOR_REBUILD_PROBE_POLICY).trim().toUpperCase()) + : null; + int base = 0; + if (centroids != null && rebuildPolicy != RebuildProbePolicy.EXACT) { + int requested = getHintInt(hint, Hint.VECTOR_PROBE_COUNT, props.getInt( + QueryServices.VECTOR_PROBE_COUNT_ATTRIB, QueryServicesOptions.DEFAULT_VECTOR_PROBE_COUNT)); + base = requested > 0 + ? Math.min(requested, centroidCount) + : Math.max(1, (int) Math.round(Math.sqrt(centroidCount))); + int count = rebuildPolicy == RebuildProbePolicy.EXPAND + ? expandProbeCount(base, + getDouble(props, QueryServices.VECTOR_REBUILD_PROBE_FACTOR_ATTRIB, + QueryServicesOptions.DEFAULT_VECTOR_REBUILD_PROBE_FACTOR), + centroidCount) Review Comment: Acknowledged. Will fix by probing both the current and new (currently being built) generation with a dual probe strategy. And we don't need policies any more, I think so will drop them and thus the point about `EXPAND` is moot. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
