Copilot commented on code in PR #2624: URL: https://github.com/apache/phoenix/pull/2624#discussion_r4198152983
########## 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: `EXPAND` cannot compensate for a rebuild with this key layout. Migration writes each processed row under the building generation’s disjoint centroid IDs and deletes its outgoing row, while this plan loads only the active centroid model and builds ranges only for active IDs. Thus the default `EXPAND` policy misses every already-migrated row regardless of expansion factor. Either retain outgoing rows until cutover, probe both generations, or force an all-generation exact scan during migration. ########## phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexScorecard.java: ########## @@ -0,0 +1,214 @@ +/* + * 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.index.vector; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.util.EnvironmentEdgeManager; +import org.apache.phoenix.util.MetaDataUtil; +import org.apache.phoenix.util.ReadOnlyProps; +import org.apache.phoenix.util.SchemaUtil; + +/** + * Vector index scorecard reconciliation and drift detection. + * <p> + * Manages posting list population and reassignment counters in {@code SYSTEM.VECTOR_CENTROID}. + * Periodic reconciliation synchronizes inline maintenance statistics with exact index row counts. + */ +public final class VectorIndexScorecard { + + /** Maximum acceptable coefficient of variation for posting list sizes before flagging drift. */ + static final double SIZE_CV_THRESHOLD = ClusterSkewMetrics.SEVERE_SKEW_CV_THRESHOLD; + /** Maximum acceptable fraction of empty posting lists before flagging drift. */ + static final double EMPTY_CENTROID_FRACTION_THRESHOLD = 0.25; + /** Maximum acceptable reassignment rate per indexed vector between reconciliations. */ + static final double REASSIGN_RATE_THRESHOLD = 0.20; + + private VectorIndexScorecard() { + } + + /** Encapsulates drift evaluation results and statistical indicators for a generation. */ + public static final class Assessment { + private final boolean drifted; + private final String reason; + private final double skewRatio; + private final double sizeCv; + private final double emptyCentroidFraction; + private final double reassignmentRate; + private final long population; + + Assessment(boolean drifted, String reason, double skewRatio, double sizeCv, + double emptyCentroidFraction, double reassignmentRate, long population) { + this.drifted = drifted; + this.reason = reason; + this.skewRatio = skewRatio; + this.sizeCv = sizeCv; + this.emptyCentroidFraction = emptyCentroidFraction; + this.reassignmentRate = reassignmentRate; + this.population = population; + } + + /** Returns true if any drift threshold was exceeded. */ + public boolean isDrifted() { + return drifted; + } + + /** Returns formatted summary of exceeded drift thresholds, or null if within limits. */ + public String getReason() { + return reason; + } + + /** Ratio of maximum posting list size to median list size. */ + public double getSkewRatio() { + return skewRatio; + } + + public double getSizeCv() { + return sizeCv; + } + + public double getEmptyCentroidFraction() { + return emptyCentroidFraction; + } + + public double getReassignmentRate() { + return reassignmentRate; + } + + public long getPopulation() { + return population; + } + + @Override + public String toString() { + return "Assessment{drifted=" + drifted + ", reason=" + reason + ", skewRatio=" + skewRatio + + ", sizeCv=" + sizeCv + ", emptyCentroidFraction=" + emptyCentroidFraction + + ", reassignmentRate=" + reassignmentRate + ", population=" + population + "}"; + } + } + + /** + * Reconciles scorecard statistics against physical index table row counts. Resets inline + * reassignment counters and updates last reconciliation timestamp. + * @param conn connection without tenant scoping to aggregate across all tenants + * @return reconciled scorecard rows with pre-reset reassignment metrics + */ + public static List<ScorecardRow> reconcile(PhoenixConnection conn, PTable index, long generation) + throws SQLException { + String indexName = index.getName().getString(); + String centroidColumn = "\"" + MetaDataUtil.VECTOR_CENTROID_ID_COLUMN_NAME + "\""; + Map<Integer, Long> populations = new HashMap<>(); + try (Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery("SELECT " + centroidColumn + ", COUNT(*) FROM " + + SchemaUtil.getEscapedFullTableName(indexName) + " GROUP BY " + centroidColumn)) { + while (rs.next()) { + populations.put(rs.getInt(1), rs.getLong(2)); + } + } + List<ScorecardRow> before = CentroidManager.loadScorecard(conn, indexName, generation); + List<ScorecardRow> reconciled = new ArrayList<>(before.size()); + List<ScorecardRow> reset = new ArrayList<>(before.size()); + for (ScorecardRow row : before) { + long population = populations.getOrDefault(row.getCentroidId(), 0L); + reconciled.add(new ScorecardRow(row.getCentroidId(), population, row.getReassignCount())); + reset.add(new ScorecardRow(row.getCentroidId(), population, 0L)); + } + CentroidManager.persistScorecard(conn, indexName, generation, reset); Review Comment: This absolute reset races with RegionServer `ON DUPLICATE KEY UPDATE` flushes. A flush committed after `loadScorecard` but before this upsert is overwritten with zero, permanently losing reassignment events (and potentially replacing newer population deltas). Reconciliation needs an atomic counter handoff/epoch or another protocol that preserves concurrent increments. ########## phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexRebuilder.java: ########## @@ -0,0 +1,445 @@ +/* + * 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.index.vector; + +import static org.apache.phoenix.query.QueryConstants.TRUE; + +import java.io.IOException; +import java.sql.SQLException; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.ResultScanner; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.filter.KeyOnlyFilter; +import org.apache.phoenix.cache.VectorCentroidCache; +import org.apache.phoenix.cache.VectorCentroidCache.CachedCentroids; +import org.apache.phoenix.compile.MutationPlan; +import org.apache.phoenix.compile.ServerBuildIndexCompiler; +import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants; +import org.apache.phoenix.exception.SQLExceptionCode; +import org.apache.phoenix.exception.SQLExceptionInfo; +import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.optimize.DistanceMetric; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTable.TaskType; +import org.apache.phoenix.schema.TableNotFoundException; +import org.apache.phoenix.util.ClientUtil; +import org.apache.phoenix.util.EnvironmentEdgeManager; +import org.apache.phoenix.util.ReadOnlyProps; +import org.apache.phoenix.util.SchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Generational rebuild orchestration and drift reconciliation for IVF vector indexes. + * <p> + * Rebuilding trains a new centroid generation, marks the index in migrating state via + * {@code SYSTEM.CATALOG}, and performs an online migration across centroid generations. Concurrent + * writes target the building generation while purging prior index rows under either generation. + * Online migration executes via a server side index build followed by catch up scan repair of + * concurrent mutations, atomic generation promotion, and retirement of prior generations. + * <p> + * Distributed mutual exclusion is maintained using atomic lease rows in + * {@code SYSTEM.VECTOR_CENTROID}. Interrupted rebuilds preserve migration state to enable + * idempotent resumption during periodic reconciliation tasks. + */ +public final class VectorIndexRebuilder { + + private static final Logger LOGGER = LoggerFactory.getLogger(VectorIndexRebuilder.class); + + /** Trigger reason identifier for operator initiated index rebuilds. */ + public static final String MANUAL_REASON = "MANUAL"; + /** Maximum lease duration for rebuild locks before allowing preemption. */ + static final long CLAIM_EXPIRY_MS = 24L * 60 * 60 * 1000; + /** + * Maximum threshold of concurrently changed rows for point catch-up repair before falling back to + * full scan. + */ + static final int MAX_CATCH_UP_ROWS = 10000; + + /** Status outcome of a rebuild execution. */ + public enum Outcome { + /** Index successfully migrated to new centroid generation. */ + REBUILT, + /** Scorecard metrics within acceptable drift tolerances. */ + NOT_DRIFTED, + /** Drift detected but automatic background rebuild is disabled. */ + DISABLED, + /** Drift detected but throttled by minimum rebuild interval. */ + TOO_SOON, + /** Rebuild or reconciliation lock currently held by another worker. */ + IN_PROGRESS, + /** Index is uninitialized or has insufficient sample vectors for training. */ + UNTRAINED, + /** Target index table not found. */ + NOT_FOUND + } + + /** Status outcome of a scorecard reconciliation execution. */ + public enum ReconcileOutcome { + /** Scorecard row counts successfully reconciled. */ + RECONCILED, + /** Reconciliation skipped due to interval throttling, uninitialized state, or active lock. */ + NOT_DUE, + /** Target index dropped; orphaned centroid metadata purged. */ + INDEX_DROPPED + } + + /** + * Test lifecycle hook invoked during migration while the building generation is populated prior + * to catalog promotion. + */ + interface MigrationHook { + void migrating(PTable index, long buildingGeneration) throws Exception; + } + + private static volatile MigrationHook migrationHookForTesting; + + /** Injects test hook for synchronization during mid-migration state. */ + static void setMigrationHookForTesting(MigrationHook hook) { + migrationHookForTesting = hook; + } + + private VectorIndexRebuilder() { + } + + /** + * Executes or resumes a vector index rebuild. Manual rebuilds bypass drift evaluation and + * throttle intervals. Automatic rebuilds verify drift, configuration flags, and minimum + * intervals. + * @param conn Phoenix connection providing execution services + * @param indexName fully qualified physical index table name + * @param manual whether this rebuild was explicitly triggered by user or operator action + * @param reason diagnostic reason or drift trigger description + */ + public static Outcome rebuild(PhoenixConnection conn, String indexName, boolean manual, + String reason) throws SQLException { + try (PhoenixConnection internal = CentroidManager.newInternalConnection(conn)) { + PTable index; + PTable dataTable; + try { + index = internal.getTableNoCache(indexName); + dataTable = internal.getTableNoCache(index.getParentName().getString()); + } catch (TableNotFoundException e) { + return Outcome.NOT_FOUND; + } + if (index.getVectorCentroidGeneration() == null) { + return Outcome.UNTRAINED; + } + String token = UUID.randomUUID().toString(); + if (!CentroidManager.claimRebuild(internal, indexName, token, CLAIM_EXPIRY_MS)) { + return Outcome.IN_PROGRESS; + } + try { + return rebuild(internal, index, dataTable, manual, reason); + } finally { + CentroidManager.releaseRebuild(internal, indexName, token); + } + } + } + + private static Outcome rebuild(PhoenixConnection conn, PTable index, PTable dataTable, + boolean manual, String reason) throws SQLException { + String indexName = index.getName().getString(); + long active = index.getVectorCentroidGeneration(); + ReadOnlyProps props = conn.getQueryServices().getProps(); + GenerationSummary activeSummary = + CentroidManager.loadGenerationSummary(conn, indexName, active); + long building; + if ( + index.isVectorRebuildInProgress() && !CentroidManager + .loadCentroids(conn, indexName, index.getVectorBuildingGeneration()).isEmpty() + ) { + building = index.getVectorBuildingGeneration(); + LOGGER.info("Resuming the migration of vector index {} to generation {}", indexName, + building); + } else { + VectorIndexScorecard.Assessment assessment = + VectorIndexScorecard.assess(VectorIndexScorecard.reconcile(conn, index, active), props); + CentroidManager.persistGenerationSummary(conn, indexName, active, + new GenerationSummary(null, assessment.getReason(), null, null, null, null)); + if (!manual) { + if (!assessment.isDrifted()) { + return Outcome.NOT_DRIFTED; Review Comment: Automatic rebuild tasks re-run reconciliation here even though the scheduling reconciliation already reset `REASSIGN_COUNT` to zero. If reassignment rate was the only exceeded threshold, this second assessment is no longer drifted and the queued task returns `NOT_DRIFTED`, so that drift trigger can never start a rebuild. Preserve/use the queued assessment reason or avoid resetting the counter before the task’s decision. ########## phoenix-core-server/src/main/java/org/apache/phoenix/iterate/VectorDistanceOrderedResultIterator.java: ########## @@ -0,0 +1,87 @@ +/* + * 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.iterate; + +import java.util.List; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.phoenix.expression.Expression; +import org.apache.phoenix.expression.OrderByExpression; +import org.apache.phoenix.expression.function.DistanceFunction; +import org.apache.phoenix.schema.SortOrder; +import org.apache.phoenix.schema.tuple.Tuple; +import org.apache.phoenix.schema.types.PDouble; + +import org.apache.phoenix.thirdparty.com.google.common.collect.MinMaxPriorityQueue; + +/** + * Server side top-N result iterator optimizing single ascending distance ordering via bounded + * evaluation. Tracks candidate pool distances of size {@code max(limit, ceil(limit * oversample))} + * and supplies the worst candidate distance as an upper bound to the distance function, enabling + * early termination during component accumulation. Candidates exceeding the upper bound are pruned + * from the sort queue. + */ +public class VectorDistanceOrderedResultIterator extends OrderedResultIterator { + + private final DistanceFunction distance; + private final MinMaxPriorityQueue<Double> pool; + private final int poolSize; + + public VectorDistanceOrderedResultIterator(ResultIterator delegate, + List<OrderByExpression> orderByExpressions, boolean spoolingEnabled, long thresholdBytes, + int limit, int estimatedRowSize, long pageSizeMs, Scan scan, RegionInfo regionInfo, + double oversampleFactor) { + super(delegate, orderByExpressions, spoolingEnabled, thresholdBytes, limit, null, + estimatedRowSize, pageSizeMs, scan, regionInfo); Review Comment: The oversampling factor only sizes the separate distance-bound queue; the `OrderedResultIterator` superclass still receives `limit`, so its result queue discards every candidate beyond K. Consequently `OVERSAMPLE` never increases the per-region candidate set returned for the global merge. Pass the computed pool size as the superclass limit as well. ########## phoenix-core-server/src/main/java/org/apache/phoenix/index/vector/ScorecardAccumulator.java: ########## @@ -0,0 +1,279 @@ +/* + * 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.index.vector; + +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CENTROID_ID; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CENTROID_VECTOR; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CLUSTER_SIZE; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GENERATION_ID; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_NAME; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REASSIGN_COUNT; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_VECTOR_CENTROID_NAME; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.client.Delete; +import org.apache.hadoop.hbase.client.Mutation; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.phoenix.hbase.index.metrics.MetricsIndexerSourceFactory; +import org.apache.phoenix.hbase.index.metrics.MetricsVectorIndexSource; +import org.apache.phoenix.index.IndexMaintainer; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; +import org.apache.phoenix.util.EnvironmentEdgeManager; +import org.apache.phoenix.util.QueryUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.phoenix.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder; + +/** + * RegionServer-level accumulator for inline vector index scorecard delta tracking. + * <p> + * Evaluates pre- and post-mutation row states to maintain accurate posting list populations and + * centroid reassignment frequencies during write operations. Scorecard deltas are buffered in + * memory and flushed periodically to {@code SYSTEM.VECTOR_CENTROID} via atomic increment upserts. + */ +public final class ScorecardAccumulator { + + private static final Logger LOGGER = LoggerFactory.getLogger(ScorecardAccumulator.class); + + private static volatile ScorecardAccumulator instance; + + private final Configuration conf; + private final ConcurrentHashMap<Key, long[]> deltas = new ConcurrentHashMap<>(); + + private ScorecardAccumulator(Configuration conf) { + this.conf = conf; + long interval = conf.getLong(QueryServices.VECTOR_SCORECARD_FLUSH_INTERVAL_MS_ATTRIB, + QueryServicesOptions.DEFAULT_VECTOR_SCORECARD_FLUSH_INTERVAL_MS); + if (interval > 0) { + ScheduledExecutorService flusher = Executors.newSingleThreadScheduledExecutor( + new ThreadFactoryBuilder().setNameFormat("vector-scorecard-flush").setDaemon(true).build()); + flusher.scheduleWithFixedDelay(this::flushQuietly, interval, interval, TimeUnit.MILLISECONDS); + } + } + + /** Returns the singleton region server accumulator instance, initializing if necessary. */ + public static ScorecardAccumulator getInstance(Configuration conf) { + ScorecardAccumulator result = instance; + if (result == null) { + synchronized (ScorecardAccumulator.class) { + result = instance; + if (result == null) { + instance = result = new ScorecardAccumulator(conf); + } + } + } + return result; + } + + /** Returns the singleton region server accumulator instance, or null if uninitialized. */ + public static ScorecardAccumulator getInstance() { + return instance; + } + + /** + * Derives scorecard deltas from index mutations for a single primary data row. + * @param maintainer index maintainer for the target vector index + * @param current prior data row image, or null if row is new + * @param mutations generated index mutations for the update + * @param pending map collecting pending centroid deltas + */ + public static void collect(IndexMaintainer maintainer, Put current, + Collection<Mutation> mutations, Map<Key, long[]> pending) { + byte[] written = null; + byte[] deleted = null; + for (Mutation mutation : mutations) { + if (mutation instanceof Put) { + written = mutation.getRow(); + } + } + for (Mutation mutation : mutations) { + // Distinguish prior row key deletion from covered column updates on the new row key + if (mutation instanceof Delete && !Bytes.equals(mutation.getRow(), written)) { + deleted = mutation.getRow(); + } + } + String indexName = maintainer.getLogicalIndexName(); + long generation = maintainer.getCentroidGeneration(); + if (written != null) { + int to = maintainer.getCentroidId(written); + if (deleted != null) { + add(pending, new Key(indexName, generation, maintainer.getCentroidId(deleted)), -1, 0); + add(pending, new Key(indexName, generation, to), 1, 1); + } else if (current == null || !maintainer.hasIndexRow(current)) { + add(pending, new Key(indexName, generation, to), 1, 0); + } + } else if (deleted != null) { + add(pending, new Key(indexName, generation, maintainer.getCentroidId(deleted)), -1, 0); + } + } + + private static void add(Map<Key, long[]> pending, Key key, long size, long reassigned) { + long[] delta = pending.computeIfAbsent(key, k -> new long[2]); + delta[0] += size; + delta[1] += reassigned; + } + + /** Aggregates batch scorecard deltas and updates metrics counters upon commit. */ + public void accumulate(Map<Key, long[]> batchDeltas) { + MetricsVectorIndexSource metrics = + MetricsIndexerSourceFactory.getInstance().getMetricsVectorIndexSource(); + for (Map.Entry<Key, long[]> entry : batchDeltas.entrySet()) { + long[] delta = entry.getValue(); + if (delta[0] == 0 && delta[1] == 0) { + continue; + } + deltas.merge(entry.getKey(), delta.clone(), + (a, b) -> new long[] { a[0] + b[0], a[1] + b[1] }); + if (delta[0] > 0) { + metrics.incrementVectorCentroidAssignments(entry.getKey().indexName, delta[0]); + } + if (delta[1] > 0) { + metrics.incrementVectorCentroidReassignments(entry.getKey().indexName, delta[1]); + } + } + } + + private void flushQuietly() { + try { + flush(); + } catch (Throwable t) { + LOGGER.warn("Vector index scorecard flush failed; reconciliation will correct the counts", t); + } + } + + /** + * Flushes buffered deltas to {@code SYSTEM.VECTOR_CENTROID} using atomic counter increments. + * Deltas for dropped indexes or retired generations are dropped. + */ + public synchronized void flush() throws SQLException { + if (deltas.isEmpty()) { + return; + } + Map<Key, long[]> drained = new HashMap<>(); + for (Key key : deltas.keySet()) { + long[] delta = deltas.remove(key); + if (delta != null) { + drained.put(key, delta); + } + } Review Comment: All pending deltas are removed before opening the connection, but an exception anywhere in the write or commit path never restores them. Population can eventually be recounted, but reassignment events cannot be reconstructed, so a transient catalog failure silently disables that drift signal for the interval. Use a lossless/idempotent handoff so failed flushes remain retryable. -- 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]
