apurtell commented on code in PR #2624: URL: https://github.com/apache/phoenix/pull/2624#discussion_r4198504178
########## 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: In fact we can remove NOT_DRIFTED entirely with a little bit of logic cleanup. -- 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]
