You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ap...@apache.org on 2026/10/08 03:21:12 UTC
(phoenix) 13/17: PHOENIX-7998 Phase 13: Drift scorecarding and background rebuild
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 42752010e188b1fd87b4dfe03d19cb3ad782c02b
Author: Andrew Purtell <ap...@apache.org>
AuthorDate: Tue Oct 6 00:32:06 2026 -0700
PHOENIX-7998 Phase 13: Drift scorecarding and background rebuild
Co-Authored-By: Claude Opus 5.5 (1M context) <no...@anthropic.com>
---
.../apache/phoenix/cache/VectorCentroidCache.java | 23 +-
.../org/apache/phoenix/compile/DeleteCompiler.java | 19 +-
.../coprocessorclient/MetaDataProtocol.java | 2 +-
.../phoenix/execute/VectorIndexScanPlan.java | 68 +-
.../index/metrics/MetricsIndexerSourceFactory.java | 8 +
.../index/metrics/MetricsVectorIndexSource.java | 72 ++
.../metrics/MetricsVectorIndexSourceImpl.java | 90 +++
.../org/apache/phoenix/index/IndexMaintainer.java | 57 +-
.../phoenix/index/vector/CentroidManager.java | 350 ++++++++-
.../phoenix/index/vector/ClusterSkewMetrics.java | 5 +-
.../phoenix/index/vector/GenerationSummary.java | 87 ++
.../apache/phoenix/index/vector/ScorecardRow.java | 49 ++
.../phoenix/index/vector/VectorIndexRebuilder.java | 444 +++++++++++
.../phoenix/index/vector/VectorIndexScorecard.java | 240 ++++++
.../phoenix/index/vector/VectorIndexTrainer.java | 17 +-
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 12 +
.../phoenix/query/ConnectionQueryServicesImpl.java | 35 +-
.../org/apache/phoenix/query/QueryConstants.java | 20 +-
.../org/apache/phoenix/query/QueryServices.java | 18 +
.../apache/phoenix/query/QueryServicesOptions.java | 6 +
.../org/apache/phoenix/schema/DelegateTable.java | 5 +
.../org/apache/phoenix/schema/MetaDataClient.java | 60 +-
.../java/org/apache/phoenix/schema/PTable.java | 17 +-
.../java/org/apache/phoenix/schema/PTableImpl.java | 26 +-
phoenix-core-client/src/main/protobuf/PTable.proto | 1 +
.../src/main/protobuf/ServerCachingService.proto | 1 +
.../coprocessor/GlobalIndexRegionScanner.java | 21 +
.../phoenix/coprocessor/MetaDataEndpointImpl.java | 21 +-
.../phoenix/coprocessor/TaskRegionObserver.java | 4 +
.../coprocessor/tasks/VectorIndexRebuildTask.java | 73 ++
.../tasks/VectorScorecardReconcileTask.java | 60 ++
.../phoenix/hbase/index/IndexRegionObserver.java | 64 +-
.../phoenix/index/vector/ScorecardAccumulator.java | 281 +++++++
.../phoenix/end2end/VectorCentroidTableIT.java | 94 ++-
.../phoenix/end2end/VectorIndexRebuildIT.java | 872 +++++++++++++++++++++
.../phoenix/end2end/VectorIndexTestUtil.java | 25 +-
.../vector/VectorIndexRebuilderTestHooks.java | 45 ++
.../phoenix/cache/VectorCentroidCacheTest.java | 12 +
.../phoenix/execute/VectorIndexScanPlanTest.java | 11 +
.../index/vector/VectorIndexRebuilderTest.java | 70 ++
.../index/vector/VectorIndexScorecardTest.java | 138 ++++
41 files changed, 3440 insertions(+), 83 deletions(-)
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/cache/VectorCentroidCache.java b/phoenix-core-client/src/main/java/org/apache/phoenix/cache/VectorCentroidCache.java
index f0cd67b6ce..da5db71374 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/cache/VectorCentroidCache.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/cache/VectorCentroidCache.java
@@ -153,15 +153,15 @@ public final class VectorCentroidCache {
private static CachedCentroids load(Connection conn, Key key, DistanceMetric metric)
throws SQLException {
- List<float[]> centroids = CentroidManager.loadCentroids(conn, key.indexName, key.generation);
- if (centroids.isEmpty()) {
+ CentroidManager.Model model = CentroidManager.loadModel(conn, key.indexName, key.generation);
+ if (model.getCentroids().isEmpty()) {
// Unrecorded centroid generation indicates stale index metadata
throw new SQLExceptionInfo.Builder(SQLExceptionCode.STALE_METADATA_CACHE_EXCEPTION)
.setMessage("No centroids recorded for vector index " + key.indexName + " generation "
+ key.generation)
.build().buildException();
}
- return new CachedCentroids(centroids, metric);
+ return new CachedCentroids(model.getCentroids(), metric, model.getFirstId());
}
private static final class Key {
@@ -199,14 +199,25 @@ public final class VectorCentroidCache {
private final float[][] centroids;
private final int dimension;
private final DistanceMetric metric;
+ private final int firstId;
public CachedCentroids(List<float[]> centroids, DistanceMetric metric) {
+ this(centroids, metric, 0);
+ }
+
+ public CachedCentroids(List<float[]> centroids, DistanceMetric metric, int firstId) {
if (centroids.isEmpty()) {
throw new IllegalArgumentException("centroids must not be empty");
}
this.centroids = centroids.toArray(new float[0][]);
this.dimension = this.centroids[0].length;
this.metric = Objects.requireNonNull(metric, "metric");
+ this.firstId = firstId;
+ }
+
+ /** Returns the initial centroid identifier for this generation. */
+ public int getFirstId() {
+ return firstId;
}
public int size() {
@@ -223,7 +234,7 @@ public final class VectorCentroidCache {
/** Returns the centroid vector at the specified index. */
public float[] getCentroid(int id) {
- return centroids[id];
+ return centroids[id - firstId];
}
/** Returns the ID of the centroid nearest to the given vector. */
@@ -238,7 +249,7 @@ public final class VectorCentroidCache {
best = c;
}
}
- return best;
+ return firstId + best;
}
/**
@@ -267,7 +278,7 @@ public final class VectorCentroidCache {
}
int[] result = new int[count];
for (int i = count - 1; i >= 0; i--) {
- result[i] = farthestFirst.poll();
+ result[i] = firstId + farthestFirst.poll();
}
return result;
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
index b07dfed0d5..abd5da9018 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
@@ -274,20 +274,24 @@ public class DeleteCompiler {
ImmutableBytesPtr otherRowKeyPtr = new ImmutableBytesPtr(); // allocate new as this is a
// key in a Map
// Translate the data table row to the index table row
+ byte[] outgoingRowKey = null;
if (table.getType() == PTableType.INDEX) {
otherRowKeyPtr.set(scannedIndexMaintainer.buildDataRowKey(rowKeyPtr, viewConstants));
if (otherTable.getType() == PTableType.INDEX) {
- byte[] otherRowKey = maintainers[i].buildRowKey(getter, otherRowKeyPtr, null, null,
- rs.getCurrentRow().getValue(0).getTimestamp());
+ long ts = rs.getCurrentRow().getValue(0).getTimestamp();
+ outgoingRowKey = maintainers[i].buildOutgoingRowKey(getter, otherRowKeyPtr, ts, null);
+ byte[] otherRowKey =
+ maintainers[i].buildRowKey(getter, otherRowKeyPtr, null, null, ts);
if (otherRowKey == null) {
continue;
}
otherRowKeyPtr.set(otherRowKey);
}
} else {
+ long ts = rs.getCurrentRow().getValue(0).getTimestamp();
+ outgoingRowKey = maintainers[i].buildOutgoingRowKey(getter, rowKeyPtr, ts, null);
// Null vectors omit index rows
- byte[] otherRowKey = maintainers[i].buildRowKey(getter, rowKeyPtr, null, null,
- rs.getCurrentRow().getValue(0).getTimestamp());
+ byte[] otherRowKey = maintainers[i].buildRowKey(getter, rowKeyPtr, null, null, ts);
if (otherRowKey == null) {
continue;
}
@@ -297,6 +301,13 @@ public class DeleteCompiler {
new RowMutationState(PRow.DELETE_MARKER, 0,
statement.getConnection().getStatementExecutionCounter(), NULL_ROWTIMESTAMP_INFO,
null));
+ if (outgoingRowKey != null && !Bytes.equals(outgoingRowKey, otherRowKeyPtr.copyBytes())) {
+ // Delete outgoing generation index entries during active migrations
+ otherMutations.get(i).put(new ImmutableBytesPtr(outgoingRowKey),
+ new RowMutationState(PRow.DELETE_MARKER, 0,
+ statement.getConnection().getStatementExecutionCounter(), NULL_ROWTIMESTAMP_INFO,
+ null));
+ }
}
if (mutations.size() > maxSize) {
throw new IllegalArgumentException("MutationState size of " + mutations.size()
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java
index 53ca5968aa..278c28ae9b 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java
@@ -92,7 +92,7 @@ public abstract class MetaDataProtocol extends MetaDataService {
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_1_0 = MIN_SYSTEM_TABLE_TIMESTAMP_4_16_0;
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 = MIN_TABLE_TIMESTAMP + 38;
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_3_0 = MIN_TABLE_TIMESTAMP + 44;
- public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 = MIN_TABLE_TIMESTAMP + 51;
+ public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 = MIN_TABLE_TIMESTAMP + 52;
// MIN_SYSTEM_TABLE_TIMESTAMP needs to be set to the max of all the MIN_SYSTEM_TABLE_TIMESTAMP_*
// constants
public static final long MIN_SYSTEM_TABLE_TIMESTAMP = MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0;
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 1d4da8bfd0..4b40e2ce39 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
@@ -92,6 +92,8 @@ public class VectorIndexScanPlan extends ScanPlan {
private final int[] probeCentroids;
private final int maxProbeBatches;
private final CachedCentroids centroids;
+ // Building generation centroids while a rebuild migration is in progress, otherwise null
+ private final CachedCentroids buildingCentroids;
private final float[] queryVector;
// Active posting list skip scan filter, updated across successive adaptive probing batches
private final SkipScanFilter postingListFilter;
@@ -109,22 +111,32 @@ public class VectorIndexScanPlan extends ScanPlan {
this.queryVector = getQueryVector(orderBy);
CachedCentroids centroids = null;
+ CachedCentroids building = null;
if (queryVector != null && index.getVectorCentroidGeneration() != null) {
- centroids = VectorCentroidCache.getInstance(connection.getQueryServices().getConfiguration())
- .get(connection, index.getName().getString(), index.getVectorCentroidGeneration(),
- DistanceMetric.fromString(index.getVectorDistanceMetric()));
+ VectorCentroidCache cache =
+ VectorCentroidCache.getInstance(connection.getQueryServices().getConfiguration());
+ DistanceMetric metric = DistanceMetric.fromString(index.getVectorDistanceMetric());
+ String indexName = index.getName().getString();
+ centroids = cache.get(connection, indexName, index.getVectorCentroidGeneration(), metric);
+ // A migration moves each index row from the active generation's posting lists to the
+ // building generation's, whose centroid IDs are disjoint, so every row is under exactly one
+ // of the two models until promotion and both must be probed
+ if (index.isVectorRebuildInProgress()) {
+ building = cache.get(connection, indexName, index.getVectorBuildingGeneration(), metric);
+ }
}
this.centroids = centroids;
- this.centroidCount = centroids == null ? 0 : centroids.size();
+ this.buildingCentroids = building;
+ this.centroidCount =
+ (centroids == null ? 0 : centroids.size()) + (building == null ? 0 : building.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));
- int count = requested > 0
- ? Math.min(requested, centroidCount)
- : Math.max(1, (int) Math.round(Math.sqrt(centroidCount)));
- int[] nearest = centroids.nearest(queryVector, count);
+ int count =
+ requested > 0 ? requested : Math.max(1, (int) Math.round(Math.sqrt(centroids.size())));
+ int[] nearest = nearestCentroids(count);
ScanRanges ranges = getPostingListRanges(context, index, nearest);
if (ranges != null) {
probes = nearest;
@@ -217,6 +229,37 @@ public class VectorIndexScanPlan extends ScanPlan {
index.getBucketNum(), true, -1);
}
+ /**
+ * Returns the {@code n} centroids nearest the query vector in each probed generation, in
+ * ascending distance order. During a migration the two generations' rankings are interleaved, so
+ * any prefix of twice a per-generation count holds that count from each model.
+ */
+ private int[] nearestCentroids(int n) {
+ int[] active = centroids.nearest(queryVector, Math.min(n, centroids.size()));
+ if (buildingCentroids == null) {
+ return active;
+ }
+ return interleave(active,
+ buildingCentroids.nearest(queryVector, Math.min(n, buildingCentroids.size())));
+ }
+
+ /** Alternates the elements of two arrays, appending the remainder of the longer one. */
+ static int[] interleave(int[] a, int[] b) {
+ int[] out = new int[a.length + b.length];
+ int i = 0;
+ int j = 0;
+ int k = 0;
+ while (i < a.length || j < b.length) {
+ if (i < a.length) {
+ out[k++] = a[i++];
+ }
+ if (j < b.length) {
+ out[k++] = b[j++];
+ }
+ }
+ return out;
+ }
+
static int getHintInt(HintNode hint, Hint name, int defaultValue) {
String value = hint.getHint(name);
if (value != null) {
@@ -274,13 +317,16 @@ public class VectorIndexScanPlan extends ScanPlan {
@Override
public ExplainPlan getExplainPlan() throws SQLException {
ExplainPlan explainPlan = super.getExplainPlan();
+ String metric = getTableRef().getTable().getVectorDistanceMetric();
+ List<String> steps = new ArrayList<>(explainPlan.getPlanSteps());
if (probeCount == 0) {
return explainPlan;
}
- String metric = getTableRef().getTable().getVectorDistanceMetric();
- List<String> steps = new ArrayList<>(explainPlan.getPlanSteps());
String probing =
"CLIENT PROBING " + probeCount + " OF " + centroidCount + " CENTROIDS (" + metric + ")";
+ if (buildingCentroids != null) {
+ probing += " ACROSS ACTIVE AND BUILDING GENERATIONS";
+ }
if (isAdaptive()) {
int batches = Math.min(maxProbeBatches, (centroidCount + probeCount - 1) / probeCount);
probing += " EXPANDING UP TO " + batches + " BATCHES";
@@ -402,7 +448,7 @@ public class VectorIndexScanPlan extends ScanPlan {
found.size() < batchLimit && probed.size() < centroidCount && batches < maxProbeBatches
) {
if (ranked == null) {
- ranked = centroids.nearest(queryVector, centroidCount);
+ ranked = nearestCentroids(centroidCount);
}
List<Integer> batch = new ArrayList<>(probeCount);
for (int i = 0; i < ranked.length && batch.size() < probeCount; i++) {
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsIndexerSourceFactory.java b/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsIndexerSourceFactory.java
index 55b648c53a..98ae3cb373 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsIndexerSourceFactory.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsIndexerSourceFactory.java
@@ -25,6 +25,7 @@ public class MetricsIndexerSourceFactory {
private volatile MetricsIndexerSource indexerSource;
private GlobalIndexCheckerSource globalIndexCheckerSource;
private MetricsIndexCDCConsumerSource indexCDCConsumerSource;
+ private MetricsVectorIndexSource vectorIndexSource;
private MetricsIndexerSourceFactory() {
}
@@ -53,4 +54,11 @@ public class MetricsIndexerSourceFactory {
}
return INSTANCE.indexCDCConsumerSource;
}
+
+ public synchronized MetricsVectorIndexSource getMetricsVectorIndexSource() {
+ if (INSTANCE.vectorIndexSource == null) {
+ INSTANCE.vectorIndexSource = new MetricsVectorIndexSourceImpl();
+ }
+ return INSTANCE.vectorIndexSource;
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsVectorIndexSource.java b/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsVectorIndexSource.java
new file mode 100644
index 0000000000..8b036a1350
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsVectorIndexSource.java
@@ -0,0 +1,72 @@
+/*
+ * 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.hbase.index.metrics;
+
+import org.apache.hadoop.hbase.metrics.BaseSource;
+
+public interface MetricsVectorIndexSource extends BaseSource {
+ // Metrics2 and JMX constants
+ String METRICS_NAME = "VectorIndex";
+ String METRICS_CONTEXT = "phoenix";
+ String METRICS_DESCRIPTION = "Metrics about Phoenix Vector Index Operations";
+ String METRICS_JMX_CONTEXT = "RegionServer,sub=" + METRICS_NAME;
+
+ String VECTOR_CENTROID_ASSIGNMENTS = "vectorCentroidAssignments";
+ String VECTOR_CENTROID_ASSIGNMENTS_DESC =
+ "The number of vectors written to a centroid's posting list by index maintenance";
+
+ String VECTOR_CENTROID_REASSIGNMENTS = "vectorCentroidReassignments";
+ String VECTOR_CENTROID_REASSIGNMENTS_DESC =
+ "The number of updated vectors index maintenance moved to a different centroid";
+
+ String VECTOR_SCORECARD_FLUSH_TIME = "vectorScorecardFlushTime";
+ String VECTOR_SCORECARD_FLUSH_TIME_DESC =
+ "Histogram for the time in milliseconds for vector scorecard flushes";
+
+ String VECTOR_SCORECARD_RECONCILE_TIME = "vectorScorecardReconcileTime";
+ String VECTOR_SCORECARD_RECONCILE_TIME_DESC =
+ "Histogram for the time in milliseconds for vector scorecard reconciliations";
+
+ /**
+ * Increments the count of vector centroid assignments.
+ * @param indexName Name of the index
+ * @param count Number of assignments
+ */
+ void incrementVectorCentroidAssignments(String indexName, long count);
+
+ /**
+ * Increments the count of vector centroid reassignments.
+ * @param indexName Name of the index
+ * @param count Number of reassignments
+ */
+ void incrementVectorCentroidReassignments(String indexName, long count);
+
+ /**
+ * Updates the scorecard flush duration histogram.
+ * @param indexName Name of the index
+ * @param timeMs Time taken in milliseconds
+ */
+ void updateVectorScorecardFlushTime(String indexName, long timeMs);
+
+ /**
+ * Updates the scorecard reconciliation duration histogram.
+ * @param indexName Name of the index
+ * @param timeMs Time taken in milliseconds
+ */
+ void updateVectorScorecardReconcileTime(String indexName, long timeMs);
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsVectorIndexSourceImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsVectorIndexSourceImpl.java
new file mode 100644
index 0000000000..515dbee447
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsVectorIndexSourceImpl.java
@@ -0,0 +1,90 @@
+/*
+ * 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.hbase.index.metrics;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.MetricHistogram;
+import org.apache.hadoop.metrics2.lib.MutableFastCounter;
+
+public class MetricsVectorIndexSourceImpl extends BaseSourceImpl
+ implements MetricsVectorIndexSource {
+
+ private final MutableFastCounter vectorCentroidAssignments;
+ private final MutableFastCounter vectorCentroidReassignments;
+ private final MetricHistogram vectorScorecardFlushTimeHisto;
+ private final MetricHistogram vectorScorecardReconcileTimeHisto;
+
+ public MetricsVectorIndexSourceImpl() {
+ this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+ }
+
+ public MetricsVectorIndexSourceImpl(String metricsName, String metricsDescription,
+ String metricsContext, String metricsJmxContext) {
+ super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+
+ vectorCentroidAssignments = getMetricsRegistry().newCounter(VECTOR_CENTROID_ASSIGNMENTS,
+ VECTOR_CENTROID_ASSIGNMENTS_DESC, 0L);
+ vectorCentroidReassignments = getMetricsRegistry().newCounter(VECTOR_CENTROID_REASSIGNMENTS,
+ VECTOR_CENTROID_REASSIGNMENTS_DESC, 0L);
+
+ vectorScorecardFlushTimeHisto = getMetricsRegistry().newHistogram(VECTOR_SCORECARD_FLUSH_TIME,
+ VECTOR_SCORECARD_FLUSH_TIME_DESC);
+ vectorScorecardReconcileTimeHisto = getMetricsRegistry()
+ .newHistogram(VECTOR_SCORECARD_RECONCILE_TIME, VECTOR_SCORECARD_RECONCILE_TIME_DESC);
+ }
+
+ @Override
+ public void incrementVectorCentroidAssignments(String indexName, long count) {
+ incrementIndexSpecificCounter(VECTOR_CENTROID_ASSIGNMENTS, indexName, count);
+ vectorCentroidAssignments.incr(count);
+ }
+
+ @Override
+ public void incrementVectorCentroidReassignments(String indexName, long count) {
+ incrementIndexSpecificCounter(VECTOR_CENTROID_REASSIGNMENTS, indexName, count);
+ vectorCentroidReassignments.incr(count);
+ }
+
+ @Override
+ public void updateVectorScorecardFlushTime(String indexName, long timeMs) {
+ incrementIndexSpecificHistogram(VECTOR_SCORECARD_FLUSH_TIME, indexName, timeMs);
+ vectorScorecardFlushTimeHisto.add(timeMs);
+ }
+
+ @Override
+ public void updateVectorScorecardReconcileTime(String indexName, long timeMs) {
+ incrementIndexSpecificHistogram(VECTOR_SCORECARD_RECONCILE_TIME, indexName, timeMs);
+ vectorScorecardReconcileTimeHisto.add(timeMs);
+ }
+
+ private void incrementIndexSpecificCounter(String baseCounterName, String indexName, long count) {
+ MutableFastCounter indexSpecificCounter =
+ getMetricsRegistry().getCounter(getCounterName(baseCounterName, indexName), 0L);
+ indexSpecificCounter.incr(count);
+ }
+
+ private void incrementIndexSpecificHistogram(String baseCounterName, String indexName, long t) {
+ MetricHistogram indexSpecificHistogram =
+ getMetricsRegistry().getHistogram(getCounterName(baseCounterName, indexName));
+ indexSpecificHistogram.add(t);
+ }
+
+ private String getCounterName(String baseCounterName, String indexName) {
+ return baseCounterName + "." + indexName;
+ }
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/IndexMaintainer.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
index 782be3d807..9867504291 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
@@ -491,7 +491,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
private String vectorAlgorithm;
private DistanceMetric distanceMetric;
private Long centroidGeneration;
- // Index column reference for expression derived vector values, or null if directly mapped.
+ private Long buildingGeneration;
private ColumnReference functionalVectorColumn;
protected IndexMaintainer(RowKeySchema dataRowKeySchema, boolean isDataTableSalted) {
@@ -523,6 +523,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
this.vectorAlgorithm = index.getVectorIndexAlgorithm();
this.distanceMetric = DistanceMetric.fromString(index.getVectorDistanceMetric());
this.centroidGeneration = index.getVectorCentroidGeneration();
+ this.buildingGeneration =
+ index.isVectorRebuildInProgress() ? index.getVectorBuildingGeneration() : null;
PColumn vectorColumn = VectorIndexTrainer.getIndexedVectorColumn(index);
if (IndexUtil.getDataColumnOrNull(dataTable, vectorColumn.getName().getString()) == null) {
this.functionalVectorColumn = new ColumnReference(vectorColumn.getFamilyName().getBytes(),
@@ -807,12 +809,22 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr,
byte[] regionStartKey, byte[] regionEndKey, long ts, byte[] encodedRegionName) {
return buildRowKey(valueGetter, rowKeyPtr, regionStartKey, regionEndKey, ts, encodedRegionName,
- null);
+ null, getWriteGeneration());
+ }
+
+ /** Generates index row key for the outgoing centroid generation during active migrations. */
+ public byte[] buildOutgoingRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr,
+ long ts, byte[] encodedRegionName) {
+ if (!isMigrating()) {
+ return null;
+ }
+ return buildRowKey(valueGetter, rowKeyPtr, null, null, ts, encodedRegionName, null,
+ centroidGeneration);
}
private byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr,
byte[] regionStartKey, byte[] regionEndKey, long ts, byte[] encodedRegionName,
- ImmutableBytesWritable indexedVector) {
+ ImmutableBytesWritable indexedVector, Long generation) {
if (isCDCIndex && encodedRegionName == null) {
throw new IllegalArgumentException("Encoded region name is required for a CDC index");
}
@@ -899,7 +911,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
if (vector == null) {
return null;
}
- ptr.set(PInteger.INSTANCE.toBytes(getCentroids().assign(vector.get(),
+ ptr.set(PInteger.INSTANCE.toBytes(getCentroids(generation).assign(vector.get(),
vector.getOffset(), vector.getLength(), dataColumnType)));
dataColumnType = PInteger.INSTANCE;
dataSortOrder = SortOrder.ASC;
@@ -1413,7 +1425,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
}
byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey,
- ts, encodedRegionName, indexedVector);
+ ts, encodedRegionName, indexedVector, getWriteGeneration());
if (indexRowKey == null) {
return null;
}
@@ -2216,6 +2228,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
maintainer.distanceMetric = DistanceMetric.fromString(proto.getDistanceMetric());
maintainer.centroidGeneration =
proto.hasCentroidGeneration() ? proto.getCentroidGeneration() : null;
+ maintainer.buildingGeneration =
+ proto.hasBuildingGeneration() ? proto.getBuildingGeneration() : null;
if (proto.hasFunctionalVectorColumn()) {
maintainer.functionalVectorColumn =
new ColumnReference(proto.getFunctionalVectorColumn().getFamily().toByteArray(),
@@ -2387,6 +2401,9 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
if (maintainer.centroidGeneration != null) {
builder.setCentroidGeneration(maintainer.centroidGeneration);
}
+ if (maintainer.buildingGeneration != null) {
+ builder.setBuildingGeneration(maintainer.buildingGeneration);
+ }
if (maintainer.functionalVectorColumn != null) {
builder.setFunctionalVectorColumn(ServerCachingProtos.ColumnReference.newBuilder()
.setFamily(ByteStringer.wrap(maintainer.functionalVectorColumn.getFamily()))
@@ -2792,6 +2809,25 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
return centroidGeneration;
}
+ /** Indicates whether an index rebuild migration is currently in progress. */
+ public boolean isMigrating() {
+ return buildingGeneration != null;
+ }
+
+ /** Returns the target centroid generation for new index row writes. */
+ public Long getWriteGeneration() {
+ return isMigrating() ? buildingGeneration : centroidGeneration;
+ }
+
+ /** Extracts the centroid identifier from an index row key. */
+ public int getCentroidId(byte[] indexRowKey) {
+ int slot =
+ (nIndexSaltBuckets > 0 ? 1 : 0) + (viewIndexId != null ? 1 : 0) + (isMultiTenant ? 1 : 0);
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ getIndexRowKeySchema().iterator(indexRowKey, ptr, slot + 1);
+ return PInteger.INSTANCE.getCodec().decodeInt(ptr, SortOrder.ASC);
+ }
+
/** Returns the column reference for computed vector expressions, or null. */
public ColumnReference getFunctionalVectorColumn() {
return functionalVectorColumn;
@@ -2804,13 +2840,15 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
public void loadCentroids(Connection conn) throws SQLException {
if (isVectorIndex()) {
VectorCentroidCache.getForWrite(conn, logicalIndexName, centroidGeneration, distanceMetric);
+ if (isMigrating()) {
+ VectorCentroidCache.getForWrite(conn, logicalIndexName, buildingGeneration, distanceMetric);
+ }
}
}
- private CachedCentroids getCentroids() {
+ private CachedCentroids getCentroids(long generation) {
try {
- return VectorCentroidCache.getForWrite(null, logicalIndexName, centroidGeneration,
- distanceMetric);
+ return VectorCentroidCache.getForWrite(null, logicalIndexName, generation, distanceMetric);
} catch (SQLException e) {
throw new IllegalStateException(e);
}
@@ -2856,7 +2894,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
* indicating no index row key relocation is required.
*/
public boolean isVectorUnchanged(Put currentDataRowState, Put nextDataRowState) {
- if (!isVectorIndex()) {
+ if (!isVectorIndex() || isMigrating()) {
+ // Migration requires rewriting unchanged vector rows under the building generation
return false;
}
ImmutableBytesWritable current = getIndexedVector(currentDataRowState, false);
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/CentroidManager.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/CentroidManager.java
index 3188239eb6..fcd92573ba 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/CentroidManager.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/CentroidManager.java
@@ -19,18 +19,28 @@ 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.INDEX_STATE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_REBUILD_TIME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_SCORECARD_UPDATE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REASSIGN_COUNT;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REBUILD_STATE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REQUESTED_LISTS;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SKEW_METRICS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_VECTOR_CENTROID_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SCHEM;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TENANT_ID;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRIGGER_REASON;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_BUILDING_GENERATION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_CENTROID_GENERATION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_IVF_LISTS;
+import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
@@ -44,12 +54,18 @@ import org.apache.phoenix.coprocessorclient.MetaDataProtocol.MutationCode;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTable.TaskStatus;
+import org.apache.phoenix.schema.PTable.TaskType;
import org.apache.phoenix.schema.TableNotFoundException;
+import org.apache.phoenix.schema.task.SystemTaskParams;
+import org.apache.phoenix.schema.task.Task;
import org.apache.phoenix.schema.types.PVectorFloat;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TaskMetaDataServiceCallBack;
/**
* Manages persistence of IVF centroid generations in {@code SYSTEM.VECTOR_CENTROID} and active
@@ -61,6 +77,11 @@ import org.apache.phoenix.util.SchemaUtil;
*/
public final class CentroidManager {
+ /** Sentinel centroid ID for generation metadata summary rows. */
+ public static final int SENTINEL_CENTROID_ID = -1;
+ /** Reserved generation ID for atomic index rebuild claim locks. */
+ static final long REBUILD_CLAIM_GENERATION = 0L;
+
private CentroidManager() {
}
@@ -94,13 +115,22 @@ public final class CentroidManager {
/** Persists centroid vectors for the specified generation and commits the transaction. */
public static void persistCentroids(Connection conn, String indexName, long generation,
List<float[]> centroids) throws SQLException {
+ persistCentroids(conn, indexName, generation, centroids, 0);
+ }
+
+ /**
+ * Persists trained centroid vectors for a generation with monotonically increasing centroid IDs
+ * starting at {@code firstId}.
+ */
+ public static void persistCentroids(Connection conn, String indexName, long generation,
+ List<float[]> centroids, int firstId) throws SQLException {
String sql = "UPSERT INTO " + SYSTEM_VECTOR_CENTROID_NAME + " (" + INDEX_NAME + ", "
+ GENERATION_ID + ", " + CENTROID_ID + ", " + CENTROID_VECTOR + ") VALUES (?, ?, ?, ?)";
try (PreparedStatement ps = conn.prepareStatement(sql)) {
for (int i = 0; i < centroids.size(); i++) {
ps.setString(1, indexName);
ps.setLong(2, generation);
- ps.setInt(3, i);
+ ps.setInt(3, firstId + i);
ps.setBytes(4, PVectorFloat.INSTANCE.toBytes(centroids.get(i)));
ps.executeUpdate();
}
@@ -108,26 +138,56 @@ public final class CentroidManager {
conn.commit();
}
+ /** Encapsulates trained centroid vectors and their starting identifier for a generation. */
+ public static final class Model {
+ private final int firstId;
+ private final List<float[]> centroids;
+
+ Model(int firstId, List<float[]> centroids) {
+ this.firstId = firstId;
+ this.centroids = centroids;
+ }
+
+ public int getFirstId() {
+ return firstId;
+ }
+
+ public List<float[]> getCentroids() {
+ return centroids;
+ }
+ }
+
/**
* Loads centroid vectors for the specified generation in ascending centroid ID order.
*/
public static List<float[]> loadCentroids(Connection conn, String indexName, long generation)
throws SQLException {
- String sql = "SELECT " + CENTROID_VECTOR + " FROM " + SYSTEM_VECTOR_CENTROID_NAME + " WHERE "
- + INDEX_NAME + " = ? AND " + GENERATION_ID + " = ? AND " + CENTROID_ID + " >= 0 AND "
- + CENTROID_VECTOR + " IS NOT NULL ORDER BY " + CENTROID_ID;
+ return loadModel(conn, indexName, generation).getCentroids();
+ }
+
+ /** Loads the centroid model and starting ID for the specified generation. */
+ public static Model loadModel(Connection conn, String indexName, long generation)
+ throws SQLException {
+ String sql =
+ "SELECT " + CENTROID_VECTOR + ", " + CENTROID_ID + " FROM " + SYSTEM_VECTOR_CENTROID_NAME
+ + " WHERE " + INDEX_NAME + " = ? AND " + GENERATION_ID + " = ? AND " + CENTROID_ID
+ + " >= 0 AND " + CENTROID_VECTOR + " IS NOT NULL ORDER BY " + CENTROID_ID;
List<float[]> centroids = new ArrayList<>();
+ int firstId = 0;
try (PreparedStatement ps = conn.prepareStatement(sql)) {
ps.setString(1, indexName);
ps.setLong(2, generation);
try (ResultSet rs = ps.executeQuery()) {
while (rs.next()) {
+ if (centroids.isEmpty()) {
+ firstId = rs.getInt(2);
+ }
byte[] b = rs.getBytes(1);
centroids.add(PVectorFloat.readElements(b, 0, b.length));
}
}
}
- return centroids;
+ return new Model(firstId, centroids);
}
/**
@@ -137,22 +197,51 @@ public final class CentroidManager {
*/
public static PTable setGenerationAndLists(PhoenixConnection conn, PTable index, long generation,
int lists) throws SQLException {
+ return recordGenerations(conn, index, generation, lists, null);
+ }
+
+ /**
+ * Updates {@code SYSTEM.CATALOG} with the target building generation for an active index rebuild.
+ */
+ public static PTable setBuildingGeneration(PhoenixConnection conn, PTable index, long generation)
+ throws SQLException {
+ return recordGenerations(conn, index, null, null, generation);
+ }
+
+ private static PTable recordGenerations(PhoenixConnection conn, PTable index, Long active,
+ Integer lists, Long building) throws SQLException {
String schemaName = index.getSchemaName().getString();
String tableName = index.getTableName().getString();
+ StringBuilder columns =
+ new StringBuilder(TENANT_ID + "," + TABLE_SCHEM + "," + TABLE_NAME + "," + INDEX_STATE);
+ StringBuilder values = new StringBuilder("?, ?, ?, ?");
+ if (active != null) {
+ columns.append(",").append(VECTOR_CENTROID_GENERATION).append(",").append(VECTOR_IVF_LISTS);
+ values.append(", ?, ?");
+ }
+ if (building != null) {
+ columns.append(",").append(VECTOR_BUILDING_GENERATION);
+ values.append(", ?");
+ }
String sql = "UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"("
- + TENANT_ID + "," + TABLE_SCHEM + "," + TABLE_NAME + "," + INDEX_STATE + ","
- + VECTOR_CENTROID_GENERATION + "," + VECTOR_IVF_LISTS + ") VALUES (?, ?, ?, ?, ?, ?)";
+ + columns + ") VALUES (" + values + ")";
boolean autoCommit = conn.getAutoCommit();
List<Mutation> tableMetadata;
conn.setAutoCommit(false);
try {
try (PreparedStatement ps = conn.prepareStatement(sql)) {
- ps.setString(1, index.getTenantId() == null ? null : index.getTenantId().getString());
- ps.setString(2, schemaName.isEmpty() ? null : schemaName);
- ps.setString(3, tableName);
- ps.setString(4, index.getIndexState().getSerializedValue());
- ps.setLong(5, generation);
- ps.setInt(6, lists);
+ int i = 1;
+ ps.setString(i++, index.getTenantId() == null ? null : index.getTenantId().getString());
+ ps.setString(i++, schemaName.isEmpty() ? null : schemaName);
+ ps.setString(i++, tableName);
+ ps.setString(i++, index.getIndexState().getSerializedValue());
+ if (active != null) {
+ ps.setLong(i++, active);
+ ps.setInt(i++, lists);
+ }
+ if (building != null) {
+ ps.setLong(i++, building);
+ }
ps.execute();
}
tableMetadata = conn.getMutationState().toMutations().next().getSecond();
@@ -167,14 +256,227 @@ public final class CentroidManager {
}
if (result.getMutationCode() != MutationCode.TABLE_ALREADY_EXISTS) {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.INVALID_INDEX_STATE_TRANSITION)
- .setMessage(
- "Could not record centroid generation " + generation + ": " + result.getMutationCode())
+ .setMessage("Could not record centroid generation " + (active != null ? active : building)
+ + ": " + result.getMutationCode())
.setSchemaName(schemaName).setTableName(tableName).build().buildException();
}
return result.getTable();
}
- /** Deletes all centroid rows for a specific generation. */
+ /**
+ * Upserts non-null generation metadata attributes into the sentinel row.
+ */
+ public static void persistGenerationSummary(Connection conn, String indexName, long generation,
+ GenerationSummary summary) throws SQLException {
+ List<String> columns = new ArrayList<>();
+ List<Object> values = new ArrayList<>();
+ addIfNotNull(columns, values, REBUILD_STATE, summary.getRebuildState());
+ addIfNotNull(columns, values, TRIGGER_REASON, summary.getTriggerReason());
+ addIfNotNull(columns, values, REQUESTED_LISTS, summary.getRequestedLists());
+ addIfNotNull(columns, values, SKEW_METRICS,
+ summary.getSkewMetrics() == null ? null : summary.getSkewMetrics().toBytes());
+ addIfNotNull(columns, values, LAST_REBUILD_TIME, summary.getLastRebuildTime());
+ addIfNotNull(columns, values, LAST_SCORECARD_UPDATE, summary.getLastScorecardUpdate());
+ StringBuilder sql = new StringBuilder("UPSERT INTO " + SYSTEM_VECTOR_CENTROID_NAME + " ("
+ + INDEX_NAME + ", " + GENERATION_ID + ", " + CENTROID_ID);
+ for (String column : columns) {
+ sql.append(", ").append(column);
+ }
+ sql.append(") VALUES (?, ?, ?");
+ for (int i = 0; i < columns.size(); i++) {
+ sql.append(", ?");
+ }
+ sql.append(")");
+ try (PreparedStatement ps = conn.prepareStatement(sql.toString())) {
+ ps.setString(1, indexName);
+ ps.setLong(2, generation);
+ ps.setInt(3, SENTINEL_CENTROID_ID);
+ for (int i = 0; i < values.size(); i++) {
+ ps.setObject(i + 4, values.get(i));
+ }
+ ps.executeUpdate();
+ }
+ conn.commit();
+ }
+
+ private static void addIfNotNull(List<String> columns, List<Object> values, String column,
+ Object value) {
+ if (value != null) {
+ columns.add(column);
+ values.add(value);
+ }
+ }
+
+ /** Reads the summary of one generation, or returns null if none was written. */
+ public static GenerationSummary loadGenerationSummary(Connection conn, String indexName,
+ long generation) throws SQLException {
+ String sql = "SELECT " + REBUILD_STATE + ", " + TRIGGER_REASON + ", " + REQUESTED_LISTS + ", "
+ + SKEW_METRICS + ", " + LAST_REBUILD_TIME + ", " + LAST_SCORECARD_UPDATE + " FROM "
+ + SYSTEM_VECTOR_CENTROID_NAME + " WHERE " + INDEX_NAME + " = ? AND " + GENERATION_ID
+ + " = ? AND " + CENTROID_ID + " = " + SENTINEL_CENTROID_ID;
+ try (PreparedStatement ps = conn.prepareStatement(sql)) {
+ ps.setString(1, indexName);
+ ps.setLong(2, generation);
+ try (ResultSet rs = ps.executeQuery()) {
+ if (!rs.next()) {
+ return null;
+ }
+ byte[] skew = rs.getBytes(4);
+ return new GenerationSummary(rs.getString(1), rs.getString(2), (Integer) rs.getObject(3),
+ skew == null ? null : ClusterSkewMetrics.fromBytes(skew), (Long) rs.getObject(5),
+ (Long) rs.getObject(6));
+ }
+ }
+ }
+
+ /**
+ * Loads scorecard statistics for the specified generation ordered by centroid ID.
+ */
+ public static List<ScorecardRow> loadScorecard(Connection conn, String indexName, long generation)
+ throws SQLException {
+ String sql = "SELECT " + CENTROID_ID + ", " + CLUSTER_SIZE + ", " + REASSIGN_COUNT + " FROM "
+ + SYSTEM_VECTOR_CENTROID_NAME + " WHERE " + INDEX_NAME + " = ? AND " + GENERATION_ID
+ + " = ? AND " + CENTROID_ID + " >= 0 AND " + CENTROID_VECTOR + " IS NOT NULL ORDER BY "
+ + CENTROID_ID;
+ List<ScorecardRow> rows = new ArrayList<>();
+ try (PreparedStatement ps = conn.prepareStatement(sql)) {
+ ps.setString(1, indexName);
+ ps.setLong(2, generation);
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ rows.add(new ScorecardRow(rs.getInt(1), rs.getLong(2), rs.getLong(3)));
+ }
+ }
+ }
+ return rows;
+ }
+
+ /**
+ * Atomically adds the cluster size and reassignment count of each row, read as deltas, to the
+ * scorecard of the specified generation and commits. RegionServer flushes and reconciliation both
+ * write through this increment, so concurrent writers compose rather than overwrite one another.
+ */
+ public static void adjustScorecard(Connection conn, String indexName, long generation,
+ List<ScorecardRow> deltas) throws SQLException {
+ String sql = "UPSERT INTO " + SYSTEM_VECTOR_CENTROID_NAME + " (" + INDEX_NAME + ", "
+ + GENERATION_ID + ", " + CENTROID_ID + ", " + CLUSTER_SIZE + ", " + REASSIGN_COUNT
+ + ") VALUES (?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE " + CLUSTER_SIZE + " = COALESCE("
+ + CLUSTER_SIZE + ", 0) + ?, " + REASSIGN_COUNT + " = COALESCE(" + REASSIGN_COUNT + ", 0) + ?";
+ try (PreparedStatement ps = conn.prepareStatement(sql)) {
+ for (ScorecardRow delta : deltas) {
+ ps.setString(1, indexName);
+ ps.setLong(2, generation);
+ ps.setInt(3, delta.getCentroidId());
+ ps.setLong(4, delta.getClusterSize());
+ ps.setLong(5, delta.getReassignCount());
+ ps.setLong(6, delta.getClusterSize());
+ ps.setLong(7, delta.getReassignCount());
+ ps.executeUpdate();
+ }
+ }
+ conn.commit();
+ }
+
+ /** Lists distinct generation IDs present in {@code SYSTEM.VECTOR_CENTROID}. */
+ public static List<Long> listGenerations(Connection conn, String indexName) throws SQLException {
+ String sql = "SELECT DISTINCT " + GENERATION_ID + " FROM " + SYSTEM_VECTOR_CENTROID_NAME
+ + " WHERE " + INDEX_NAME + " = ? AND " + GENERATION_ID + " > " + REBUILD_CLAIM_GENERATION;
+ List<Long> generations = new ArrayList<>();
+ try (PreparedStatement ps = conn.prepareStatement(sql)) {
+ ps.setString(1, indexName);
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ generations.add(rs.getLong(1));
+ }
+ }
+ }
+ return generations;
+ }
+
+ /**
+ * Atomically acquires an exclusive rebuild lease in {@code SYSTEM.VECTOR_CENTROID}. Stale leases
+ * older than {@code expiryMs} may be preempted.
+ * @return true if the lease was successfully acquired by {@code token}
+ */
+ public static boolean claimRebuild(Connection conn, String indexName, String token, long expiryMs)
+ throws SQLException {
+ long now = EnvironmentEdgeManager.currentTimeMillis();
+ long expired = now - expiryMs;
+ String sql = "UPSERT INTO " + SYSTEM_VECTOR_CENTROID_NAME + " (" + INDEX_NAME + ", "
+ + GENERATION_ID + ", " + CENTROID_ID + ", " + TRIGGER_REASON + ", " + LAST_REBUILD_TIME
+ + ") VALUES (?, " + REBUILD_CLAIM_GENERATION + ", " + SENTINEL_CENTROID_ID
+ + ", ?, ?) ON DUPLICATE KEY UPDATE " + TRIGGER_REASON + " = CASE WHEN " + LAST_REBUILD_TIME
+ + " < ? THEN ? ELSE " + TRIGGER_REASON + " END, " + LAST_REBUILD_TIME + " = CASE WHEN "
+ + LAST_REBUILD_TIME + " < ? THEN ? ELSE " + LAST_REBUILD_TIME + " END";
+ try (PreparedStatement ps = conn.prepareStatement(sql)) {
+ ps.setString(1, indexName);
+ ps.setString(2, token);
+ ps.setLong(3, now);
+ ps.setLong(4, expired);
+ ps.setString(5, token);
+ ps.setLong(6, expired);
+ ps.setLong(7, now);
+ ps.executeUpdate();
+ }
+ conn.commit();
+ try (PreparedStatement ps = conn.prepareStatement("SELECT " + TRIGGER_REASON + " FROM "
+ + SYSTEM_VECTOR_CENTROID_NAME + " WHERE " + INDEX_NAME + " = ? AND " + GENERATION_ID + " = "
+ + REBUILD_CLAIM_GENERATION + " AND " + CENTROID_ID + " = " + SENTINEL_CENTROID_ID)) {
+ ps.setString(1, indexName);
+ try (ResultSet rs = ps.executeQuery()) {
+ return rs.next() && token.equals(rs.getString(1));
+ }
+ }
+ }
+
+ /** Releases the rebuild lease held by {@code token}. */
+ public static void releaseRebuild(Connection conn, String indexName, String token)
+ throws SQLException {
+ try (PreparedStatement ps = conn.prepareStatement("DELETE FROM " + SYSTEM_VECTOR_CENTROID_NAME
+ + " WHERE " + INDEX_NAME + " = ? AND " + GENERATION_ID + " = " + REBUILD_CLAIM_GENERATION
+ + " AND " + CENTROID_ID + " = " + SENTINEL_CENTROID_ID + " AND " + TRIGGER_REASON + " = ?")) {
+ ps.setString(1, indexName);
+ ps.setString(2, token);
+ executeAutoCommitted(conn, ps);
+ }
+ }
+
+ /**
+ * Enqueues an index task in {@code SYSTEM.TASK} if no active task of the given type exists.
+ */
+ public static void enqueueTask(PhoenixConnection conn, PTable index, TaskType taskType,
+ String data) throws SQLException {
+ String schemaName = index.getSchemaName().getString();
+ String tableName = index.getTableName().getString();
+ String tenantId = index.getTenantId() == null ? null : index.getTenantId().getString();
+ for (Task.TaskRecord task : Task.queryTaskTable(conn, null, schemaName, tableName, taskType,
+ tenantId, null)) {
+ String status = task.getStatus();
+ if (
+ !TaskStatus.COMPLETED.toString().equals(status)
+ && !TaskStatus.FAILED.toString().equals(status)
+ ) {
+ return;
+ }
+ }
+ try {
+ List<Mutation> mutations = Task.getMutationsForAddTask(
+ new SystemTaskParams.SystemTaskParamsBuilder().setConn(conn).setTaskType(taskType)
+ .setTenantId(tenantId).setSchemaName(schemaName.isEmpty() ? null : schemaName)
+ .setTableName(tableName).setData(data).build());
+ MetaDataMutationResult result = Task.taskMetaDataCoprocessorExec(conn,
+ mutations.get(0).getRow(), new TaskMetaDataServiceCallBack(mutations));
+ if (MutationCode.UNABLE_TO_UPSERT_TASK.equals(result.getMutationCode())) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.UNABLE_TO_UPSERT_TASK)
+ .setSchemaName(schemaName).setTableName(tableName).build().buildException();
+ }
+ } catch (IOException e) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.UNABLE_TO_UPSERT_TASK).setRootCause(e)
+ .setSchemaName(schemaName).setTableName(tableName).build().buildException();
+ }
+ }
+
+ /** Deletes all centroid and scorecard rows for the specified generation. */
public static void deleteGeneration(Connection conn, String indexName, long generation)
throws SQLException {
try (PreparedStatement ps = conn.prepareStatement("DELETE FROM " + SYSTEM_VECTOR_CENTROID_NAME
@@ -194,6 +496,22 @@ public final class CentroidManager {
}
}
+ /** Removes vector index rebuild and reconciliation tasks from {@code SYSTEM.TASK}. */
+ public static void deleteTasks(Connection conn, PTable index) throws SQLException {
+ String schemaName = index.getSchemaName().getString();
+ try (PreparedStatement ps = conn.prepareStatement("DELETE FROM "
+ + PhoenixDatabaseMetaData.SYSTEM_TASK_NAME + " WHERE " + PhoenixDatabaseMetaData.TASK_TYPE
+ + " IN (" + TaskType.VECTOR_SCORECARD_RECONCILE.getSerializedValue() + ", "
+ + TaskType.VECTOR_INDEX_REBUILD.getSerializedValue() + ") AND " + TABLE_NAME + " = ? AND "
+ + (schemaName.isEmpty() ? TABLE_SCHEM + " IS NULL" : TABLE_SCHEM + " = ?"))) {
+ ps.setString(1, index.getTableName().getString());
+ if (!schemaName.isEmpty()) {
+ ps.setString(2, schemaName);
+ }
+ executeAutoCommitted(conn, ps);
+ }
+ }
+
/**
* Executes a delete statement with autocommit enabled to allow server side range deletion.
*/
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/ClusterSkewMetrics.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/ClusterSkewMetrics.java
index a6844ac42f..90c7707320 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/ClusterSkewMetrics.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/ClusterSkewMetrics.java
@@ -171,11 +171,14 @@ public class ClusterSkewMetrics {
}
}
+ /** Coefficient of variation threshold indicating severe cluster distribution skew. */
+ public static final double SEVERE_SKEW_CV_THRESHOLD = 1.0;
+
/**
* Indicates whether cluster distribution exhibits severe skew (coefficient of variation > 1.0).
*/
public boolean isSevereSkew() {
- return coefficientOfVariation > 1.0;
+ return coefficientOfVariation > SEVERE_SKEW_CV_THRESHOLD;
}
public int getMin() {
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/GenerationSummary.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/GenerationSummary.java
new file mode 100644
index 0000000000..2927dce207
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/GenerationSummary.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.index.vector;
+
+/**
+ * Metadata and drift metrics for a centroid generation, persisted in the sentinel row
+ * ({@code CENTROID_ID = -1}) of {@code SYSTEM.VECTOR_CENTROID}. Null fields indicate unset values
+ * and preserve existing column values during partial updates.
+ */
+public final class GenerationSummary {
+
+ /** Generation is active for index maintenance and queries. */
+ public static final String ACTIVE = "A";
+ /** Generation is undergoing background rebuild migration. */
+ public static final String BUILDING = "B";
+
+ private final String rebuildState;
+ private final String triggerReason;
+ private final Integer requestedLists;
+ private final ClusterSkewMetrics skewMetrics;
+ private final Long lastRebuildTime;
+ private final Long lastScorecardUpdate;
+
+ public GenerationSummary(String rebuildState, String triggerReason, Integer requestedLists,
+ ClusterSkewMetrics skewMetrics, Long lastRebuildTime, Long lastScorecardUpdate) {
+ this.rebuildState = rebuildState;
+ this.triggerReason = triggerReason;
+ this.requestedLists = requestedLists;
+ this.skewMetrics = skewMetrics;
+ this.lastRebuildTime = lastRebuildTime;
+ this.lastScorecardUpdate = lastScorecardUpdate;
+ }
+
+ /** Rebuild lifecycle state ({@link #ACTIVE}, {@link #BUILDING}, or null). */
+ public String getRebuildState() {
+ return rebuildState;
+ }
+
+ /** Reason triggering initial creation or subsequent rebuild recommendation. */
+ public String getTriggerReason() {
+ return triggerReason;
+ }
+
+ /**
+ * Configured target IVF list count before split heuristic adjustments.
+ */
+ public Integer getRequestedLists() {
+ return requestedLists;
+ }
+
+ /** Training sample cluster skew distribution metrics. */
+ public ClusterSkewMetrics getSkewMetrics() {
+ return skewMetrics;
+ }
+
+ /** Timestamp when this generation was promoted to active. */
+ public Long getLastRebuildTime() {
+ return lastRebuildTime;
+ }
+
+ /** Timestamp of the most recent scorecard reconciliation. */
+ public Long getLastScorecardUpdate() {
+ return lastScorecardUpdate;
+ }
+
+ @Override
+ public String toString() {
+ return "GenerationSummary{rebuildState=" + rebuildState + ", triggerReason=" + triggerReason
+ + ", requestedLists=" + requestedLists + ", skewMetrics=" + skewMetrics + ", lastRebuildTime="
+ + lastRebuildTime + ", lastScorecardUpdate=" + lastScorecardUpdate + "}";
+ }
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/ScorecardRow.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/ScorecardRow.java
new file mode 100644
index 0000000000..ea150e4fb6
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/ScorecardRow.java
@@ -0,0 +1,49 @@
+/*
+ * 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;
+
+/** Centroid drift scorecard statistics tracking cluster population and reassignment counts. */
+public final class ScorecardRow {
+ private final int centroidId;
+ private final long clusterSize;
+ private final long reassignCount;
+
+ public ScorecardRow(int centroidId, long clusterSize, long reassignCount) {
+ this.centroidId = centroidId;
+ this.clusterSize = clusterSize;
+ this.reassignCount = reassignCount;
+ }
+
+ public int getCentroidId() {
+ return centroidId;
+ }
+
+ public long getClusterSize() {
+ return clusterSize;
+ }
+
+ public long getReassignCount() {
+ return reassignCount;
+ }
+
+ @Override
+ public String toString() {
+ return "ScorecardRow{centroidId=" + centroidId + ", clusterSize=" + clusterSize
+ + ", reassignCount=" + reassignCount + "}";
+ }
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexRebuilder.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexRebuilder.java
new file mode 100644
index 0000000000..20b5f2b268
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexRebuilder.java
@@ -0,0 +1,444 @@
+/*
+ * 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,
+ /** 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 the automatic rebuild flag
+ * and throttle interval. Automatic rebuilds are queued by {@link #reconcile} once it has assessed
+ * the index as drifted, and recheck only the automatic rebuild flag and minimum interval.
+ * @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 for an automatic rebuild the drift assessment that
+ * queued it
+ */
+ 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 {
+ if (manual) {
+ 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));
+ } else {
+ // Drift was assessed by the reconciliation that queued this rebuild, over freshly
+ // reconciled counts. Reconciling again would consume the reassignment counters that
+ // assessment was based on, so only the operational gates are rechecked here.
+ if (
+ !props.getBoolean(QueryServices.VECTOR_REBUILD_AUTO_ENABLED_ATTRIB,
+ QueryServicesOptions.DEFAULT_VECTOR_REBUILD_AUTO_ENABLED)
+ ) {
+ return Outcome.DISABLED;
+ }
+ // Generation ID timestamp bounds the minimum elapsed time since previous rebuild
+ long lastRebuild = activeSummary != null && activeSummary.getLastRebuildTime() != null
+ ? Math.max(activeSummary.getLastRebuildTime(), active)
+ : active;
+ if (
+ EnvironmentEdgeManager.currentTimeMillis() - lastRebuild
+ < props.getLong(QueryServices.VECTOR_REBUILD_MIN_INTERVAL_MS_ATTRIB,
+ QueryServicesOptions.DEFAULT_VECTOR_REBUILD_MIN_INTERVAL_MS)
+ ) {
+ return Outcome.TOO_SOON;
+ }
+ }
+ Integer requestedLists = activeSummary != null ? activeSummary.getRequestedLists() : null;
+ int lists = requestedLists != null ? requestedLists : index.getVectorIvfLists();
+ KMeansResult model = VectorIndexTrainer.train(conn, dataTable, index, lists);
+ if (model == null) {
+ return Outcome.UNTRAINED;
+ }
+ building = CentroidManager.nextGeneration(active);
+ // Assign centroid IDs past outgoing ranges to avoid row key collisions with prior delete
+ // markers
+ CachedCentroids outgoing =
+ VectorCentroidCache.getInstance(conn.getQueryServices().getConfiguration()).get(conn,
+ indexName, active, DistanceMetric.fromString(index.getVectorDistanceMetric()));
+ int firstId = outgoing.getFirstId() + outgoing.size();
+ CentroidManager.persistCentroids(conn, indexName, building, model.getCentroids(), firstId);
+ CentroidManager.persistGenerationSummary(conn, indexName, building, new GenerationSummary(
+ GenerationSummary.BUILDING, reason, lists, model.getSkewMetrics(), null, null));
+ VectorCentroidCache.getInstance(conn.getQueryServices().getConfiguration()).put(indexName,
+ building, new CachedCentroids(model.getCentroids(), model.getDistanceMetric(), firstId));
+ CentroidManager.setBuildingGeneration(conn, index, building);
+ LOGGER.info("Migrating vector index {} from generation {} to generation {}: {}", indexName,
+ active, building, reason);
+ }
+ // Wait for client metadata caches to observe building generation
+ awaitMetadataRefresh(conn);
+ index = refresh(conn, index);
+ migrate(conn, index, dataTable, building);
+
+ List<float[]> centroids = CentroidManager.loadCentroids(conn, indexName, building);
+ CentroidManager.setGenerationAndLists(conn, index, building, centroids.size());
+ CentroidManager.persistGenerationSummary(conn, indexName, building,
+ new GenerationSummary(GenerationSummary.ACTIVE, null, null, null,
+ EnvironmentEdgeManager.currentTimeMillis(), null));
+ // Wait for client metadata caches to observe promoted active generation
+ awaitMetadataRefresh(conn);
+ index = refresh(conn, index);
+ retireInactiveGenerations(conn, index);
+ VectorIndexScorecard.reconcile(conn, index, building);
+ LOGGER.info("Rebuilt vector index {} to generation {}", indexName, building);
+ return Outcome.REBUILT;
+ }
+
+ /**
+ * Executes dual phase migration, a full server side index rebuild followed by catch up scan for
+ * concurrent modifications.
+ */
+ private static void migrate(PhoenixConnection conn, PTable index, PTable dataTable, long building)
+ throws SQLException {
+ String dataTableName = SchemaUtil.getEscapedFullTableName(dataTable.getName().getString());
+ long start = EnvironmentEdgeManager.currentTimeMillis();
+ MutationPlan plan = new ServerBuildIndexCompiler(conn, dataTableName).compile(index);
+ setTimeRange(plan.getContext().getScan(), 0, start);
+ conn.getQueryServices().updateData(plan);
+ MigrationHook hook = migrationHookForTesting;
+ if (hook != null) {
+ try {
+ hook.migrating(index, building);
+ } catch (Exception e) {
+ throw new SQLException(e);
+ }
+ }
+ long end = EnvironmentEdgeManager.currentTimeMillis();
+ Set<ImmutableBytesPtr> changed = getChangedRows(conn, dataTable, start, end);
+ if (changed.size() > MAX_CATCH_UP_ROWS) {
+ plan = new ServerBuildIndexCompiler(conn, dataTableName).compile(index);
+ setTimeRange(plan.getContext().getScan(), 0, end);
+ conn.getQueryServices().updateData(plan);
+ } else if (!changed.isEmpty()) {
+ rebuildRows(conn, index, dataTableName, dataTable, changed, end);
+ }
+ }
+
+ /** Identifies primary data row keys modified within the specified timestamp window. */
+ private static Set<ImmutableBytesPtr> getChangedRows(PhoenixConnection conn, PTable dataTable,
+ long start, long end) throws SQLException {
+ Scan scan = new Scan();
+ scan.setRaw(true);
+ scan.setFilter(new KeyOnlyFilter());
+ setTimeRange(scan, start, end);
+ Set<ImmutableBytesPtr> rows = new HashSet<>();
+ try (Table table = conn.getQueryServices().getTable(dataTable.getPhysicalName().getBytes());
+ ResultScanner scanner = table.getScanner(scan)) {
+ for (Result result = scanner.next(); result != null; result = scanner.next()) {
+ rows.add(new ImmutableBytesPtr(result.getRow()));
+ if (rows.size() > MAX_CATCH_UP_ROWS) {
+ break;
+ }
+ }
+ } catch (IOException e) {
+ throw ClientUtil.parseServerException(e);
+ }
+ return rows;
+ }
+
+ /** Rebuilds index rows for specific primary keys via point scans. */
+ private static void rebuildRows(PhoenixConnection conn, PTable index, String dataTableName,
+ PTable dataTable, Set<ImmutableBytesPtr> rows, long end) throws SQLException {
+ Scan template =
+ new ServerBuildIndexCompiler(conn, dataTableName).compile(index).getContext().getScan();
+ template.setAttribute(BaseScannerRegionObserverConstants.UNGROUPED_AGG, TRUE);
+ try (Table table = conn.getQueryServices().getTable(dataTable.getPhysicalName().getBytes())) {
+ for (ImmutableBytesPtr row : rows) {
+ Scan scan = new Scan(template);
+ scan.withStartRow(row.copyBytesIfNecessary(), true);
+ scan.withStopRow(row.copyBytesIfNecessary(), true);
+ setTimeRange(scan, 0, end);
+ try (ResultScanner scanner = table.getScanner(scan)) {
+ scanner.next();
+ }
+ }
+ } catch (IOException e) {
+ throw ClientUtil.parseServerException(e);
+ }
+ }
+
+ /**
+ * Reconciles scorecard statistics, evaluates drift conditions, purges retired generations, and
+ * schedules background rebuild tasks as necessary.
+ * @param conn Phoenix connection providing execution services
+ * @param indexName fully qualified physical index table name
+ */
+ public static ReconcileOutcome reconcile(PhoenixConnection conn, String indexName)
+ throws SQLException {
+ try (PhoenixConnection internal = CentroidManager.newInternalConnection(conn)) {
+ PTable index;
+ try {
+ index = internal.getTableNoCache(indexName);
+ } catch (TableNotFoundException e) {
+ // Clean up orphaned centroid records following concurrent table drop
+ CentroidManager.deleteAllCentroids(internal, indexName);
+ return ReconcileOutcome.INDEX_DROPPED;
+ }
+ Long active = index.getVectorCentroidGeneration();
+ if (active == null) {
+ return ReconcileOutcome.NOT_DUE;
+ }
+ ReadOnlyProps props = internal.getQueryServices().getProps();
+ GenerationSummary summary =
+ CentroidManager.loadGenerationSummary(internal, indexName, active);
+ boolean due = isDue(summary == null ? null : summary.getLastScorecardUpdate(),
+ props.getLong(QueryServices.VECTOR_SCORECARD_RECONCILE_INTERVAL_MS_ATTRIB,
+ QueryServicesOptions.DEFAULT_VECTOR_SCORECARD_RECONCILE_INTERVAL_MS),
+ EnvironmentEdgeManager.currentTimeMillis());
+ List<Long> generations = CentroidManager.listGenerations(internal, indexName);
+ boolean retire = generations.size() > (index.isVectorRebuildInProgress() ? 2 : 1);
+ if (!due && !retire && !index.isVectorRebuildInProgress()) {
+ return ReconcileOutcome.NOT_DUE;
+ }
+ String token = UUID.randomUUID().toString();
+ if (!CentroidManager.claimRebuild(internal, indexName, token, CLAIM_EXPIRY_MS)) {
+ return ReconcileOutcome.NOT_DUE;
+ }
+ try {
+ retireInactiveGenerations(internal, index);
+ if (index.isVectorRebuildInProgress()) {
+ CentroidManager.enqueueTask(internal, index, TaskType.VECTOR_INDEX_REBUILD,
+ rebuildTaskData(true, "RESUME"));
+ return ReconcileOutcome.NOT_DUE;
+ }
+ if (!due) {
+ return ReconcileOutcome.NOT_DUE;
+ }
+ VectorIndexScorecard.Assessment assessment = VectorIndexScorecard
+ .assess(VectorIndexScorecard.reconcile(internal, index, active), props);
+ CentroidManager.persistGenerationSummary(internal, indexName, active,
+ new GenerationSummary(null, assessment.getReason(), null, null, null, null));
+ if (
+ assessment.isDrifted()
+ && props.getBoolean(QueryServices.VECTOR_REBUILD_AUTO_ENABLED_ATTRIB,
+ QueryServicesOptions.DEFAULT_VECTOR_REBUILD_AUTO_ENABLED)
+ ) {
+ CentroidManager.enqueueTask(internal, index, TaskType.VECTOR_INDEX_REBUILD,
+ rebuildTaskData(false, assessment.getReason()));
+ }
+ return ReconcileOutcome.RECONCILED;
+ } finally {
+ CentroidManager.releaseRebuild(internal, indexName, token);
+ }
+ }
+ }
+
+ /** Evaluates whether reconciliation is due based on elapsed time since previous update. */
+ static boolean isDue(Long lastUpdate, long intervalMs, long now) {
+ if (lastUpdate == null) {
+ return true;
+ }
+ long elapsed = now - lastUpdate;
+ return elapsed >= 0 && elapsed >= intervalMs;
+ }
+
+ /** Serializes task payload parameters for background rebuild tasks. */
+ public static String rebuildTaskData(boolean manual, String reason) {
+ return "{\"manual\":" + manual + ",\"reason\":\""
+ + (reason == null ? "" : reason.replace("\"", "'")) + "\"}";
+ }
+
+ /** Purges centroid rows and metadata for retired generations. */
+ private static void retireInactiveGenerations(PhoenixConnection conn, PTable index)
+ throws SQLException {
+ String indexName = index.getName().getString();
+ for (long generation : CentroidManager.listGenerations(conn, indexName)) {
+ if (
+ !Long.valueOf(generation).equals(index.getVectorCentroidGeneration())
+ && !Long.valueOf(generation).equals(index.getVectorBuildingGeneration())
+ ) {
+ CentroidManager.deleteGeneration(conn, indexName, generation);
+ }
+ }
+ }
+
+ /** Blocks execution to allow distributed client metadata cache invalidation. */
+ private static void awaitMetadataRefresh(PhoenixConnection conn) throws SQLException {
+ long sleep =
+ conn.getQueryServices().getProps().getLong(QueryServices.INDEX_POPULATION_SLEEP_TIME,
+ QueryServicesOptions.DEFAULT_INDEX_POPULATION_SLEEP_TIME);
+ if (sleep > 0) {
+ try {
+ Thread.sleep(sleep);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.INTERRUPTED_EXCEPTION).setRootCause(e)
+ .build().buildException();
+ }
+ }
+ }
+
+ private static PTable refresh(PhoenixConnection conn, PTable index) throws SQLException {
+ String indexName = index.getName().getString();
+ conn.removeTable(null, indexName, index.getParentName().getString(),
+ org.apache.hadoop.hbase.HConstants.LATEST_TIMESTAMP);
+ return conn.getTableNoCache(indexName);
+ }
+
+ private static void setTimeRange(Scan scan, long min, long max) throws SQLException {
+ try {
+ scan.setTimeRange(min, max);
+ } catch (IOException e) {
+ throw new SQLException(e);
+ }
+ }
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexScorecard.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexScorecard.java
new file mode 100644
index 0000000000..a6b8ada46d
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexScorecard.java
@@ -0,0 +1,240 @@
+/*
+ * 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 + "}";
+ }
+ }
+
+ /**
+ * Test hook invoked after reconciliation reads the scorecard and before it writes corrections.
+ */
+ interface ReconcileHook {
+ void loaded(String indexName, long generation) throws Exception;
+ }
+
+ private static volatile ReconcileHook reconcileHookForTesting;
+
+ /** Injects test hook for interleaving scorecard writes with reconciliation. */
+ static void setReconcileHookForTesting(ReconcileHook hook) {
+ reconcileHookForTesting = hook;
+ }
+
+ /**
+ * Reconciles scorecard statistics against physical index table row counts. Resets inline
+ * reassignment counters and updates last reconciliation timestamp. Both are applied as
+ * corrections relative to the counters read, so concurrent RegionServer flushes are kept.
+ * @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());
+ // Corrections are relative to the values read, so RegionServer flushes committed after the
+ // read are preserved rather than overwritten
+ List<ScorecardRow> corrections = new ArrayList<>(before.size());
+ for (ScorecardRow row : before) {
+ long population = populations.getOrDefault(row.getCentroidId(), 0L);
+ reconciled.add(new ScorecardRow(row.getCentroidId(), population, row.getReassignCount()));
+ corrections.add(new ScorecardRow(row.getCentroidId(), population - row.getClusterSize(),
+ -row.getReassignCount()));
+ }
+ ReconcileHook hook = reconcileHookForTesting;
+ if (hook != null) {
+ try {
+ hook.loaded(indexName, generation);
+ } catch (Exception e) {
+ throw new SQLException(e);
+ }
+ }
+ CentroidManager.adjustScorecard(conn, indexName, generation, corrections);
+ CentroidManager.persistGenerationSummary(conn, indexName, generation, new GenerationSummary(
+ null, null, null, null, null, EnvironmentEdgeManager.currentTimeMillis()));
+ return reconciled;
+ }
+
+ /**
+ * Evaluates scorecard metrics against configured drift thresholds (skew ratio, size CV, empty
+ * fraction, and reassignment rate). Bypassed when total population is below minimum.
+ */
+ public static Assessment assess(List<ScorecardRow> rows, ReadOnlyProps props) {
+ int k = rows.size();
+ if (k == 0) {
+ return new Assessment(false, null, 0, 0, 0, 0, 0);
+ }
+ long[] sizes = new long[k];
+ long population = 0;
+ long reassigned = 0;
+ int empty = 0;
+ for (int i = 0; i < k; i++) {
+ ScorecardRow row = rows.get(i);
+ sizes[i] = row.getClusterSize();
+ population += sizes[i];
+ reassigned += row.getReassignCount();
+ if (sizes[i] == 0) {
+ empty++;
+ }
+ }
+ Arrays.sort(sizes);
+ long max = sizes[k - 1];
+ double median = k % 2 == 1 ? sizes[k / 2] : (sizes[k / 2 - 1] + sizes[k / 2]) / 2.0;
+ // Skew relative to median cluster size.
+ // Non-zero maximum with zero median indicates unbounded skew
+ double skewRatio = median > 0 ? max / median : max > 0 ? Double.POSITIVE_INFINITY : 0;
+ double mean = (double) population / k;
+ double sumSquares = 0;
+ for (long size : sizes) {
+ sumSquares += (size - mean) * (size - mean);
+ }
+ double sizeCv = mean > 0 ? Math.sqrt(sumSquares / k) / mean : 0;
+ double emptyFraction = (double) empty / k;
+ double reassignmentRate = population > 0 ? (double) reassigned / population : 0;
+
+ long minPopulation = props.getLong(QueryServices.VECTOR_DRIFT_MIN_POPULATION_ATTRIB,
+ QueryServicesOptions.DEFAULT_VECTOR_DRIFT_MIN_POPULATION);
+ if (population < minPopulation) {
+ return new Assessment(false, null, skewRatio, sizeCv, emptyFraction, reassignmentRate,
+ population);
+ }
+ double skewThreshold =
+ Double.parseDouble(props.get(QueryServices.VECTOR_DRIFT_SKEW_RATIO_THRESHOLD_ATTRIB,
+ Double.toString(QueryServicesOptions.DEFAULT_VECTOR_DRIFT_SKEW_RATIO_THRESHOLD)));
+ List<String> reasons = new ArrayList<>(4);
+ addIfExceeded(reasons, "SKEW_RATIO_EXCEEDED", skewRatio, skewThreshold);
+ addIfExceeded(reasons, "SIZE_CV_EXCEEDED", sizeCv, SIZE_CV_THRESHOLD);
+ addIfExceeded(reasons, "EMPTY_CENTROID_FRACTION_EXCEEDED", emptyFraction,
+ EMPTY_CENTROID_FRACTION_THRESHOLD);
+ addIfExceeded(reasons, "REASSIGN_RATE_EXCEEDED", reassignmentRate, REASSIGN_RATE_THRESHOLD);
+ return new Assessment(!reasons.isEmpty(), reasons.isEmpty() ? null : String.join("; ", reasons),
+ skewRatio, sizeCv, emptyFraction, reassignmentRate, population);
+ }
+
+ private static void addIfExceeded(List<String> reasons, String name, double value,
+ double threshold) {
+ if (value > threshold) {
+ reasons.add(name + ": " + (Double.isInfinite(value) ? "unbounded" : format(value)) + " > "
+ + format(threshold));
+ }
+ }
+
+ private static String format(double value) {
+ return String.format("%.2f", value);
+ }
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexTrainer.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexTrainer.java
index 7ffaf653b5..695cd49a11 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexTrainer.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexTrainer.java
@@ -46,6 +46,8 @@ public final class VectorIndexTrainer {
private static final Logger LOGGER = LoggerFactory.getLogger(VectorIndexTrainer.class);
/** Oversampling margin to ensure sufficient sample volume for the reservoir. */
private static final double OVERSAMPLE = 1.2;
+ /** Initial creation trigger reason for newly trained vector indexes. */
+ public static final String CREATE_INDEX_REASON = "CREATE_INDEX";
private VectorIndexTrainer() {
}
@@ -68,7 +70,17 @@ public final class VectorIndexTrainer {
/** Samples base table vectors and trains centroids according to the index configuration. */
public static KMeansResult train(PhoenixConnection conn, PTable dataTable, PTable index)
throws SQLException {
- int lists = index.getVectorIvfLists();
+ return train(conn, dataTable, index, index.getVectorIvfLists());
+ }
+
+ /**
+ * Samples base table vectors and trains the requested number of centroids using the index
+ * configuration.
+ * @return trained k-means model, or null if sample volume is insufficient for the requested
+ * clusters
+ */
+ public static KMeansResult train(PhoenixConnection conn, PTable dataTable, PTable index,
+ int lists) throws SQLException {
int sampleSize = index.getVectorIvfSampleSize();
DistanceMetric metric = DistanceMetric.fromString(index.getVectorDistanceMetric());
PColumn vectorColumn = getIndexedVectorColumn(index);
@@ -112,6 +124,9 @@ public final class VectorIndexTrainer {
String indexName = index.getName().getString();
long generation = CentroidManager.nextGeneration(index.getVectorCentroidGeneration());
CentroidManager.persistCentroids(conn, indexName, generation, result.getCentroids());
+ CentroidManager.persistGenerationSummary(conn, indexName, generation,
+ new GenerationSummary(GenerationSummary.ACTIVE, CREATE_INDEX_REASON, result.getRequestedK(),
+ result.getSkewMetrics(), null, null));
CentroidManager.setGenerationAndLists(conn, index, generation, result.getEffectiveK());
VectorCentroidCache.getInstance(conn.getQueryServices().getConfiguration()).put(indexName,
generation, new CachedCentroids(result.getCentroids(), result.getDistanceMetric()));
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 399fd9a88f..373420fee5 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -470,6 +470,10 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final byte[] VECTOR_CENTROID_GENERATION_BYTES =
Bytes.toBytes(VECTOR_CENTROID_GENERATION);
+ public static final String VECTOR_BUILDING_GENERATION = "VECTOR_BUILDING_GENERATION";
+ public static final byte[] VECTOR_BUILDING_GENERATION_BYTES =
+ Bytes.toBytes(VECTOR_BUILDING_GENERATION);
+
public static final String SYSTEM_CHILD_LINK_TABLE = "CHILD_LINK";
public static final String SYSTEM_CHILD_LINK_NAME =
SchemaUtil.getTableName(SYSTEM_CATALOG_SCHEMA, SYSTEM_CHILD_LINK_TABLE);
@@ -544,6 +548,14 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final byte[] CENTROID_VECTOR_BYTES = Bytes.toBytes(CENTROID_VECTOR);
public static final String GENERATION_ID = "GENERATION_ID";
public static final byte[] GENERATION_ID_BYTES = Bytes.toBytes(GENERATION_ID);
+ public static final String CLUSTER_SIZE = "CLUSTER_SIZE";
+ public static final String REASSIGN_COUNT = "REASSIGN_COUNT";
+ public static final String SKEW_METRICS = "SKEW_METRICS";
+ public static final String REBUILD_STATE = "REBUILD_STATE";
+ public static final String TRIGGER_REASON = "TRIGGER_REASON";
+ public static final String REQUESTED_LISTS = "REQUESTED_LISTS";
+ public static final String LAST_REBUILD_TIME = "LAST_REBUILD_TIME";
+ public static final String LAST_SCORECARD_UPDATE = "LAST_SCORECARD_UPDATE";
public static final String QUERY_ID = "QUERY_ID";
public static final String USER = "USER";
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index bd22f0625c..db4bfa3db3 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -4821,22 +4821,22 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
}
if (currentServerSideTableTimeStamp < MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0) {
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 15,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 16,
PhoenixDatabaseMetaData.PHYSICAL_TABLE_NAME + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 14,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 15,
PhoenixDatabaseMetaData.SCHEMA_VERSION + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 13,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 14,
PhoenixDatabaseMetaData.EXTERNAL_SCHEMA_ID + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 12,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 13,
PhoenixDatabaseMetaData.STREAMING_TOPIC_NAME + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 11,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 12,
PhoenixDatabaseMetaData.INDEX_WHERE + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 10,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 11,
PhoenixDatabaseMetaData.CDC_INCLUDE_TABLE + " " + PVarchar.INSTANCE.getSqlTypeName());
/**
@@ -4844,35 +4844,38 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
* PHOENIX_TTL Column. See PHOENIX-7023
*/
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 9,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 10,
PhoenixDatabaseMetaData.TTL + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 8,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 9,
PhoenixDatabaseMetaData.ROW_KEY_MATCHER + " " + PVarbinary.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 7,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 8,
PhoenixDatabaseMetaData.IS_STRICT_TTL + " " + PBoolean.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 6,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 7,
PhoenixDatabaseMetaData.INDEX_CONSISTENCY + " CHAR(1)");
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 5,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 6,
PhoenixDatabaseMetaData.VECTOR_INDEX_ALGORITHM + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 4,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 5,
PhoenixDatabaseMetaData.VECTOR_DISTANCE_METRIC + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 3,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 4,
PhoenixDatabaseMetaData.VECTOR_DIMENSION + " " + PInteger.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 2,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 3,
PhoenixDatabaseMetaData.VECTOR_IVF_LISTS + " " + PInteger.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 1,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 2,
PhoenixDatabaseMetaData.VECTOR_IVF_SAMPLE_SIZE + " " + PInteger.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 1,
PhoenixDatabaseMetaData.VECTOR_CENTROID_GENERATION + " " + PLong.INSTANCE.getSqlTypeName());
+ metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0,
+ PhoenixDatabaseMetaData.VECTOR_BUILDING_GENERATION + " " + PLong.INSTANCE.getSqlTypeName());
// move TTL values stored in descriptor to SYSCAT TTL column.
moveTTLFromHBaseLevelTTLToPhoenixLevelTTL(metaConnection);
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java
index c5f5423412..288a8ebe2e 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@ -32,6 +32,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CHANGE_DETECTION_E
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CHAR_OCTET_LENGTH;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CLASS_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CLIENT_IP;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CLUSTER_SIZE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_COUNT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_DEF;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_FAMILY;
@@ -78,6 +79,8 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_VIEW_REFERENCED
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.JAR_PATH;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.KEY_SEQ;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_DDL_TIMESTAMP;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_REBUILD_TIME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_SCORECARD_UPDATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_STATS_UPDATE_TIME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_TIMESTAMP;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG;
@@ -111,8 +114,11 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PK_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.QUERY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.QUERY_ID;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.QUERY_STATUS;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REASSIGN_COUNT;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REBUILD_STATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REF_GENERATION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REMARKS;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REQUESTED_LISTS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.RETURN_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ROW_KEY_MATCHER;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SALT_BUCKETS;
@@ -124,6 +130,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SCOPE_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SELF_REFERENCING_COL_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SEQUENCE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SEQUENCE_SCHEMA;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SKEW_METRICS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SORT_ORDER;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SOURCE_DATA_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SQL_DATA_TYPE;
@@ -171,6 +178,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSFORM_START_TS
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSFORM_STATUS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSFORM_TABLE_TTL;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSFORM_TYPE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRIGGER_REASON;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TTL;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TTL_FOR_MUTEX;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE;
@@ -179,6 +187,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_SEQUENCE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.UPDATE_CACHE_FREQUENCY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USER;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USE_STATS_FOR_PARALLELIZATION;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_BUILDING_GENERATION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_CENTROID_GENERATION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_DIMENSION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_DISTANCE_METRIC;
@@ -433,7 +442,7 @@ public interface QueryConstants {
+ IS_STRICT_TTL + " BOOLEAN, \n" + INDEX_CONSISTENCY + " CHAR(1), \n" + VECTOR_INDEX_ALGORITHM
+ " VARCHAR, \n" + VECTOR_DISTANCE_METRIC + " VARCHAR, \n" + VECTOR_DIMENSION + " INTEGER, \n"
+ VECTOR_IVF_LISTS + " INTEGER, \n" + VECTOR_IVF_SAMPLE_SIZE + " INTEGER, \n"
- + VECTOR_CENTROID_GENERATION + " BIGINT, \n" +
+ + VECTOR_CENTROID_GENERATION + " BIGINT, \n" + VECTOR_BUILDING_GENERATION + " BIGINT, \n" +
// Column metadata (will be null for table row)
DATA_TYPE + " INTEGER," + COLUMN_SIZE + " INTEGER," + DECIMAL_DIGITS + " INTEGER," + NULLABLE
+ " INTEGER," + ORDINAL_POSITION + " INTEGER," + SORT_ORDER + " INTEGER," + ARRAY_SIZE
@@ -632,8 +641,11 @@ public interface QueryConstants {
INDEX_NAME + " VARCHAR NOT NULL," + GENERATION_ID + " BIGINT NOT NULL," + CENTROID_ID
+ " INTEGER NOT NULL,\n" +
// Non-PK columns
- CENTROID_VECTOR + " VARBINARY,\n" + "CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY ("
- + INDEX_NAME + "," + GENERATION_ID + "," + CENTROID_ID + "))\n" + HConstants.VERSIONS
- + "=%s,\n" + ColumnFamilyDescriptorBuilder.KEEP_DELETED_CELLS + "=%s,\n" + TRANSACTIONAL + "="
+ CENTROID_VECTOR + " VARBINARY," + CLUSTER_SIZE + " BIGINT," + REASSIGN_COUNT + " BIGINT,"
+ + SKEW_METRICS + " VARBINARY," + REBUILD_STATE + " CHAR(1)," + TRIGGER_REASON + " VARCHAR,"
+ + REQUESTED_LISTS + " INTEGER," + LAST_REBUILD_TIME + " BIGINT," + LAST_SCORECARD_UPDATE
+ + " BIGINT,\n" + "CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + INDEX_NAME + ","
+ + GENERATION_ID + "," + CENTROID_ID + "))\n" + HConstants.VERSIONS + "=%s,\n"
+ + ColumnFamilyDescriptorBuilder.KEEP_DELETED_CELLS + "=%s,\n" + TRANSACTIONAL + "="
+ Boolean.FALSE;
}
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 3a43a518cb..26943e8c94 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
@@ -156,6 +156,24 @@ public interface QueryServices extends SQLCloseable {
public static final String VECTOR_PROBE_COUNT_ATTRIB = "phoenix.vector.probe.count";
// Maximum number of probe batches scanned during adaptive vector query execution
public static final String VECTOR_MAX_PROBE_LIMIT_ATTRIB = "phoenix.vector.max.probe.limit";
+ // Vector index scorecard flush interval in milliseconds
+ public static final String VECTOR_SCORECARD_FLUSH_INTERVAL_MS_ATTRIB =
+ "phoenix.vector.scorecard.flush.interval.ms";
+ // Vector index scorecard reconciliation interval in milliseconds
+ public static final String VECTOR_SCORECARD_RECONCILE_INTERVAL_MS_ATTRIB =
+ "phoenix.vector.scorecard.reconcile.interval.ms";
+ // Posting list skew ratio threshold relative to median list size for drift detection
+ public static final String VECTOR_DRIFT_SKEW_RATIO_THRESHOLD_ATTRIB =
+ "phoenix.vector.drift.skew.ratio.threshold";
+ // Minimum index population threshold required for drift evaluation
+ public static final String VECTOR_DRIFT_MIN_POPULATION_ATTRIB =
+ "phoenix.vector.drift.min.population";
+ // Enable automatic background rebuild for drifted vector indexes
+ public static final String VECTOR_REBUILD_AUTO_ENABLED_ATTRIB =
+ "phoenix.vector.rebuild.auto.enabled";
+ // Minimum elapsed interval between consecutive background rebuilds of a vector index
+ public static final String VECTOR_REBUILD_MIN_INTERVAL_MS_ATTRIB =
+ "phoenix.vector.rebuild.min.interval.ms";
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 cad482412f..65c519dea9 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
@@ -235,6 +235,12 @@ public class QueryServicesOptions {
public static final long DEFAULT_VECTOR_CENTROID_CACHE_MAX_SIZE = 1000L;
public static final int DEFAULT_VECTOR_PROBE_COUNT = 0;
public static final int DEFAULT_VECTOR_MAX_PROBE_LIMIT = 8;
+ public static final long DEFAULT_VECTOR_SCORECARD_FLUSH_INTERVAL_MS = 60000L;
+ public static final long DEFAULT_VECTOR_SCORECARD_RECONCILE_INTERVAL_MS = 86400000L;
+ public static final double DEFAULT_VECTOR_DRIFT_SKEW_RATIO_THRESHOLD = 4.0;
+ public static final long DEFAULT_VECTOR_DRIFT_MIN_POPULATION = 1000L;
+ public static final boolean DEFAULT_VECTOR_REBUILD_AUTO_ENABLED = false;
+ public static final long DEFAULT_VECTOR_REBUILD_MIN_INTERVAL_MS = 86400000L;
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-client/src/main/java/org/apache/phoenix/schema/DelegateTable.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/DelegateTable.java
index 8a9a3adda0..3db8a0fde1 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/DelegateTable.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/DelegateTable.java
@@ -510,4 +510,9 @@ public class DelegateTable implements PTable {
public Long getVectorCentroidGeneration() {
return delegate.getVectorCentroidGeneration();
}
+
+ @Override
+ public Long getVectorBuildingGeneration() {
+ return delegate.getVectorBuildingGeneration();
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 2754b064d5..0e78af26fc 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -229,6 +229,8 @@ import org.apache.phoenix.expression.function.PhoenixRowTimestampFunction;
import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.index.vector.CentroidManager;
+import org.apache.phoenix.index.vector.VectorIndexRebuilder;
+import org.apache.phoenix.index.vector.VectorIndexScorecard;
import org.apache.phoenix.index.vector.VectorIndexTrainer;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
@@ -1543,6 +1545,12 @@ public class MetaDataClient {
dataTableRef.getTable().getTableName().getString(), false, PIndexState.ACTIVE);
alterIndex(indexStatement);
+ if (index.isVectorIndex()) {
+ // Initialize baseline scorecard counts for index maintenance
+ try (PhoenixConnection internal = CentroidManager.newInternalConnection(connection)) {
+ VectorIndexScorecard.reconcile(internal, index, index.getVectorCentroidGeneration());
+ }
+ }
return state;
} finally {
connection.setAutoCommit(wasAutoCommit);
@@ -1987,6 +1995,10 @@ public class MetaDataClient {
if (LOGGER.isInfoEnabled())
LOGGER.info("Created index " + table.getName().getString() + " at " + table.getTimeStamp());
+ if (table.isVectorIndex()) {
+ enqueueScorecardReconciliation(table);
+ }
+
if (
statement.getIndexConsistency() != null && statement.getIndexConsistency().isAsynchronous()
) {
@@ -2035,6 +2047,19 @@ public class MetaDataClient {
return state;
}
+ /**
+ * Enqueues the periodic task responsible for vector index scorecard reconciliation.
+ */
+ private void enqueueScorecardReconciliation(PTable index) {
+ try (PhoenixConnection internal = CentroidManager.newInternalConnection(connection)) {
+ CentroidManager.enqueueTask(internal, index, PTable.TaskType.VECTOR_SCORECARD_RECONCILE,
+ null);
+ } catch (SQLException e) {
+ LOGGER.warn("Could not enqueue scorecard reconciliation for vector index {}",
+ index.getName().getString(), e);
+ }
+ }
+
/**
* Trains and persists the initial centroid generation for a newly created vector index. Training
* is deferred if non-null vector count is less than the requested list count.
@@ -4645,9 +4670,32 @@ public class MetaDataClient {
}
}
+ /** Executes a synchronous or asynchronous vector index rebuild and migration. */
+ private MutationState rebuildVectorIndex(PTable index, boolean isAsync) throws SQLException {
+ String indexName = index.getName().getString();
+ if (isAsync) {
+ try (PhoenixConnection internal = CentroidManager.newInternalConnection(connection)) {
+ CentroidManager.enqueueTask(internal, index, PTable.TaskType.VECTOR_INDEX_REBUILD,
+ VectorIndexRebuilder.rebuildTaskData(true, VectorIndexRebuilder.MANUAL_REASON));
+ }
+ return new MutationState(0, 0, connection);
+ }
+ VectorIndexRebuilder.Outcome outcome =
+ VectorIndexRebuilder.rebuild(connection, indexName, true, VectorIndexRebuilder.MANUAL_REASON);
+ if (outcome != VectorIndexRebuilder.Outcome.REBUILT) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.INVALID_INDEX_STATE_TRANSITION)
+ .setMessage("Vector index " + indexName + " was not rebuilt: " + outcome)
+ .setSchemaName(index.getSchemaName().getString())
+ .setTableName(index.getTableName().getString()).build().buildException();
+ }
+ connection.removeTable(connection.getTenantId(), indexName, index.getParentName().getString(),
+ HConstants.LATEST_TIMESTAMP);
+ return new MutationState(0, 0, connection);
+ }
+
/**
- * Deletes SYSTEM.VECTOR_CENTROID entries for a dropped vector index using an isolated connection
- * to preserve caller transaction state.
+ * Deletes centroid metadata and scheduled tasks for a dropped vector index using an isolated
+ * connection.
*/
private void deleteVectorCentroids(PTable index) {
if (!index.isVectorIndex()) {
@@ -4658,9 +4706,10 @@ public class MetaDataClient {
.invalidate(indexName);
try (PhoenixConnection internal = CentroidManager.newInternalConnection(connection)) {
CentroidManager.deleteAllCentroids(internal, indexName);
+ CentroidManager.deleteTasks(internal, index);
} catch (SQLException e) {
- LOGGER.warn("Could not delete the centroids of dropped vector index {}; its "
- + PhoenixDatabaseMetaData.SYSTEM_VECTOR_CENTROID_NAME + " rows remain.", indexName, e);
+ LOGGER.warn("Could not delete the centroids and tasks of dropped vector index {}; its "
+ + "reconciliation task removes what remains.", indexName, e);
}
}
@@ -6198,6 +6247,9 @@ public class MetaDataClient {
.setSchemaName(schemaName).setTableName(indexName).build().buildException();
}
+ if (newIndexState == PIndexState.REBUILD && table.isVectorIndex()) {
+ return rebuildVectorIndex(table, isAsync);
+ }
if (newIndexState == PIndexState.REBUILD) {
newIndexState = PIndexState.BUILDING;
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java
index 42b83fbaef..a4b7f520af 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java
@@ -208,7 +208,9 @@ public interface PTable extends PMetaDataEntity {
DROP_CHILD_VIEWS((byte) 1),
INDEX_REBUILD((byte) 2),
TRANSFORM_MONITOR((byte) 3),
- CDC_STREAM_PARTITION((byte) 4);
+ CDC_STREAM_PARTITION((byte) 4),
+ VECTOR_SCORECARD_RECONCILE((byte) 5),
+ VECTOR_INDEX_REBUILD((byte) 6);
private final byte[] byteValue;
private final byte serializedValue;
@@ -1081,6 +1083,19 @@ public interface PTable extends PMetaDataEntity {
/** Returns the active centroid generation ID, or null if not a vector index. */
Long getVectorCentroidGeneration();
+ /**
+ * Returns the target centroid generation for an active index rebuild migration, or null if
+ * inactive.
+ */
+ Long getVectorBuildingGeneration();
+
+ /** Indicates whether an index rebuild migration is currently in progress. */
+ default boolean isVectorRebuildInProgress() {
+ Long building = getVectorBuildingGeneration();
+ Long active = getVectorCentroidGeneration();
+ return building != null && (active == null || building > active);
+ }
+
/** Returns true if this table is a vector index. */
default boolean isVectorIndex() {
return getIndexType() == IndexType.VECTOR_GLOBAL;
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTableImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTableImpl.java
index 45667de0ea..5f312be419 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTableImpl.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTableImpl.java
@@ -230,6 +230,7 @@ public class PTableImpl implements PTable {
private final Integer vectorIvfLists;
private final Integer vectorIvfSampleSize;
private final Long vectorCentroidGeneration;
+ private final Long vectorBuildingGeneration;
public static class Builder {
private PTableKey key;
@@ -306,6 +307,7 @@ public class PTableImpl implements PTable {
private Integer vectorIvfLists;
private Integer vectorIvfSampleSize;
private Long vectorCentroidGeneration;
+ private Long vectorBuildingGeneration;
// Used to denote which properties a view has explicitly modified
private BitSet viewModifiedPropSet = new BitSet(3);
@@ -787,6 +789,11 @@ public class PTableImpl implements PTable {
return this;
}
+ public Builder setVectorBuildingGeneration(Long vectorBuildingGeneration) {
+ this.vectorBuildingGeneration = vectorBuildingGeneration;
+ return this;
+ }
+
/**
* Populate derivable attributes of the PTable
* @return PTableImpl.Builder object
@@ -1076,6 +1083,7 @@ public class PTableImpl implements PTable {
this.vectorIvfLists = builder.vectorIvfLists;
this.vectorIvfSampleSize = builder.vectorIvfSampleSize;
this.vectorCentroidGeneration = builder.vectorCentroidGeneration;
+ this.vectorBuildingGeneration = builder.vectorBuildingGeneration;
}
// When cloning table, ignore the salt column as it will be added back in the constructor
@@ -1153,7 +1161,8 @@ public class PTableImpl implements PTable {
.setVectorDistanceMetric(table.getVectorDistanceMetric())
.setVectorDimension(table.getVectorDimension()).setVectorIvfLists(table.getVectorIvfLists())
.setVectorIvfSampleSize(table.getVectorIvfSampleSize())
- .setVectorCentroidGeneration(table.getVectorCentroidGeneration());
+ .setVectorCentroidGeneration(table.getVectorCentroidGeneration())
+ .setVectorBuildingGeneration(table.getVectorBuildingGeneration());
}
@Override
@@ -2188,6 +2197,10 @@ public class PTableImpl implements PTable {
if (table.hasVectorCentroidGeneration()) {
vectorCentroidGeneration = table.getVectorCentroidGeneration();
}
+ Long vectorBuildingGeneration = null;
+ if (table.hasVectorBuildingGeneration()) {
+ vectorBuildingGeneration = table.getVectorBuildingGeneration();
+ }
IndexConsistency indexConsistency = null;
if (tableType == PTableType.INDEX) {
if (table.hasIndexConsistency()) {
@@ -2235,7 +2248,8 @@ public class PTableImpl implements PTable {
.setVectorIndexAlgorithm(vectorIndexAlgorithm).setVectorDistanceMetric(vectorDistanceMetric)
.setVectorDimension(vectorDimension).setVectorIvfLists(vectorIvfLists)
.setVectorIvfSampleSize(vectorIvfSampleSize)
- .setVectorCentroidGeneration(vectorCentroidGeneration).build();
+ .setVectorCentroidGeneration(vectorCentroidGeneration)
+ .setVectorBuildingGeneration(vectorBuildingGeneration).build();
} catch (SQLException e) {
throw new RuntimeException(e); // Impossible
}
@@ -2426,6 +2440,9 @@ public class PTableImpl implements PTable {
if (table.getVectorCentroidGeneration() != null) {
builder.setVectorCentroidGeneration(table.getVectorCentroidGeneration());
}
+ if (table.getVectorBuildingGeneration() != null) {
+ builder.setVectorBuildingGeneration(table.getVectorBuildingGeneration());
+ }
return builder.build();
}
@@ -2699,6 +2716,11 @@ public class PTableImpl implements PTable {
return vectorCentroidGeneration;
}
+ @Override
+ public Long getVectorBuildingGeneration() {
+ return vectorBuildingGeneration;
+ }
+
private static final class KVColumnFamilyQualifier {
@Nonnull
private final String colFamilyName;
diff --git a/phoenix-core-client/src/main/protobuf/PTable.proto b/phoenix-core-client/src/main/protobuf/PTable.proto
index d2c4496d75..8ef6d71f39 100644
--- a/phoenix-core-client/src/main/protobuf/PTable.proto
+++ b/phoenix-core-client/src/main/protobuf/PTable.proto
@@ -131,6 +131,7 @@ message PTable {
optional int32 vectorIvfLists = 62;
optional int32 vectorIvfSampleSize = 63;
optional int64 vectorCentroidGeneration = 64;
+ optional int64 vectorBuildingGeneration = 65;
}
message EncodedCQCounter {
diff --git a/phoenix-core-client/src/main/protobuf/ServerCachingService.proto b/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
index 1c719fcc2f..0c939368f3 100644
--- a/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
+++ b/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
@@ -78,6 +78,7 @@ message IndexMaintainer {
optional string distanceMetric = 35;
optional int64 centroidGeneration = 36;
optional ColumnReference functionalVectorColumn = 37;
+ optional int64 buildingGeneration = 38;
}
message TransformMaintainer {
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java
index 96c72c302d..4475e247cd 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java
@@ -1214,6 +1214,23 @@ public abstract class GlobalIndexRegionScanner extends BaseRegionScanner {
return mutationList;
}
+ /**
+ * Generates a delete mutation for the outgoing generation index entry during online migration.
+ */
+ private static void addOutgoingRowDelete(IndexMaintainer indexMaintainer, ValueGetter dataRowVG,
+ ImmutableBytesPtr rowKeyPtr, long ts, byte[] encodedRegionName, byte[] builtRowKey,
+ List<Mutation> indexMutations) {
+ if (!indexMaintainer.isMigrating()) {
+ return;
+ }
+ byte[] outgoingRowKey =
+ indexMaintainer.buildOutgoingRowKey(dataRowVG, rowKeyPtr, ts, encodedRegionName);
+ if (outgoingRowKey != null && !Bytes.equals(outgoingRowKey, builtRowKey)) {
+ indexMutations.add(indexMaintainer.buildRowDeleteMutation(outgoingRowKey,
+ IndexMaintainer.DeleteType.ALL_VERSIONS, ts));
+ }
+ }
+
private static Put prepareIndexPutForRebuild(IndexMaintainer indexMaintainer,
ImmutableBytesPtr rowKeyPtr, ValueGetter mergedRowVG, long ts, byte[] encodedRegionName)
throws IOException {
@@ -1375,6 +1392,8 @@ public abstract class GlobalIndexRegionScanner extends BaseRegionScanner {
Put indexPut = prepareIndexPutForRebuild(indexMaintainer, rowKeyPtr, nextDataRowVG, ts,
encodedRegionName);
indexMutations.add(indexPut);
+ addOutgoingRowDelete(indexMaintainer, nextDataRowVG, rowKeyPtr, ts, encodedRegionName,
+ indexPut.getRow(), indexMutations);
Delete deleteColumn = indexMaintainer.buildDeleteColumnMutation(indexPut, ts);
if (deleteColumn != null) {
indexMutations.add(deleteColumn);
@@ -1435,6 +1454,8 @@ public abstract class GlobalIndexRegionScanner extends BaseRegionScanner {
Put indexPut = prepareIndexPutForRebuild(indexMaintainer, rowKeyPtr, nextDataRowVG, ts,
encodedRegionName);
indexMutations.add(indexPut);
+ addOutgoingRowDelete(indexMaintainer, nextDataRowVG, rowKeyPtr, ts, encodedRegionName,
+ indexPut.getRow(), indexMutations);
Delete deleteColumn = indexMaintainer.buildDeleteColumnMutation(indexPut, ts);
if (deleteColumn != null) {
indexMutations.add(deleteColumn);
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index 405d47f2a6..37e43ca636 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -81,6 +81,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TTL_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.UPDATE_CACHE_FREQUENCY_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USE_STATS_FOR_PARALLELIZATION_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_BUILDING_GENERATION_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_CENTROID_GENERATION_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_DIMENSION_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_DISTANCE_METRIC_BYTES;
@@ -442,6 +443,8 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, VECTOR_IVF_SAMPLE_SIZE_BYTES);
private static final Cell VECTOR_CENTROID_GENERATION_KV = createFirstOnRow(
ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, VECTOR_CENTROID_GENERATION_BYTES);
+ private static final Cell VECTOR_BUILDING_GENERATION_KV = createFirstOnRow(
+ ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, VECTOR_BUILDING_GENERATION_BYTES);
private static final Cell TTL_KV =
createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, TTL_BYTES);
@@ -462,7 +465,7 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
SCHEMA_VERSION_KV, EXTERNAL_SCHEMA_ID_KV, STREAMING_TOPIC_NAME_KV, INDEX_WHERE_KV,
CDC_INCLUDE_KV, TTL_KV, ROW_KEY_MATCHER_KV, IS_STRICT_TTL_KV, INDEX_CONSISTENCY_KV,
VECTOR_INDEX_ALGORITHM_KV, VECTOR_DISTANCE_METRIC_KV, VECTOR_DIMENSION_KV, VECTOR_IVF_LISTS_KV,
- VECTOR_IVF_SAMPLE_SIZE_KV, VECTOR_CENTROID_GENERATION_KV);
+ VECTOR_IVF_SAMPLE_SIZE_KV, VECTOR_CENTROID_GENERATION_KV, VECTOR_BUILDING_GENERATION_KV);
static {
Collections.sort(TABLE_KV_COLUMNS, CellComparatorImpl.COMPARATOR);
@@ -535,6 +538,8 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
TABLE_KV_COLUMNS.indexOf(VECTOR_IVF_SAMPLE_SIZE_KV);
private static final int VECTOR_CENTROID_GENERATION_INDEX =
TABLE_KV_COLUMNS.indexOf(VECTOR_CENTROID_GENERATION_KV);
+ private static final int VECTOR_BUILDING_GENERATION_INDEX =
+ TABLE_KV_COLUMNS.indexOf(VECTOR_BUILDING_GENERATION_KV);
// KeyValues for Column
private static final KeyValue DECIMAL_DIGITS_KV =
createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, DECIMAL_DIGITS_BYTES);
@@ -1698,6 +1703,17 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
: oldTable != null ? oldTable.getVectorCentroidGeneration()
: null);
+ Cell vectorBuildingGenerationKv = tableKeyValues[VECTOR_BUILDING_GENERATION_INDEX];
+ Long vectorBuildingGeneration = null;
+ if (vectorBuildingGenerationKv != null) {
+ vectorBuildingGeneration =
+ (Long) PLong.INSTANCE.toObject(vectorBuildingGenerationKv.getValueArray(),
+ vectorBuildingGenerationKv.getValueOffset(), vectorBuildingGenerationKv.getValueLength());
+ }
+ builder.setVectorBuildingGeneration(vectorBuildingGeneration != null ? vectorBuildingGeneration
+ : oldTable != null ? oldTable.getVectorBuildingGeneration()
+ : null);
+
// Check the cell tag to see whether the view has modified this property
final byte[] tagUseStatsForParallelization = (useStatsForParallelizationKv == null)
? HConstants.EMPTY_BYTE_ARRAY
@@ -4613,6 +4629,9 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
Bytes.compareTo(cell.getQualifierArray(), cell.getQualifierOffset(),
cell.getQualifierLength(), VECTOR_CENTROID_GENERATION_BYTES, 0,
VECTOR_CENTROID_GENERATION_BYTES.length) == 0
+ || Bytes.compareTo(cell.getQualifierArray(), cell.getQualifierOffset(),
+ cell.getQualifierLength(), VECTOR_BUILDING_GENERATION_BYTES, 0,
+ VECTOR_BUILDING_GENERATION_BYTES.length) == 0
) {
setsVectorGeneration = true;
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/TaskRegionObserver.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/TaskRegionObserver.java
index dc6055862e..ae65b7dac5 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/TaskRegionObserver.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/TaskRegionObserver.java
@@ -76,6 +76,10 @@ public class TaskRegionObserver implements RegionObserver, RegionCoprocessor {
.put(TaskType.TRANSFORM_MONITOR, "org.apache.phoenix.coprocessor.tasks.TransformMonitorTask")
.put(TaskType.CDC_STREAM_PARTITION,
"org.apache.phoenix.coprocessor.tasks.CdcStreamPartitionMetadataTask")
+ .put(TaskType.VECTOR_SCORECARD_RECONCILE,
+ "org.apache.phoenix.coprocessor.tasks.VectorScorecardReconcileTask")
+ .put(TaskType.VECTOR_INDEX_REBUILD,
+ "org.apache.phoenix.coprocessor.tasks.VectorIndexRebuildTask")
.build();
public enum TaskResultCode {
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/VectorIndexRebuildTask.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/VectorIndexRebuildTask.java
new file mode 100644
index 0000000000..f0216b30c9
--- /dev/null
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/VectorIndexRebuildTask.java
@@ -0,0 +1,73 @@
+/*
+ * 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.coprocessor.tasks;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.phoenix.coprocessor.TaskRegionObserver.TaskResult;
+import org.apache.phoenix.coprocessor.TaskRegionObserver.TaskResultCode;
+import org.apache.phoenix.index.vector.VectorIndexRebuilder;
+import org.apache.phoenix.index.vector.VectorIndexRebuilder.Outcome;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.JacksonUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.SchemaUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.phoenix.thirdparty.com.google.common.base.Strings;
+
+/**
+ * Server side background task orchestrating asynchronous vector index rebuilds. Enqueued during
+ * drift detection or via {@code ALTER INDEX ... REBUILD ASYNC}.
+ */
+public class VectorIndexRebuildTask extends BaseTask {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(VectorIndexRebuildTask.class);
+
+ @Override
+ public TaskResult run(Task.TaskRecord taskRecord) {
+ String indexName =
+ SchemaUtil.getTableName(taskRecord.getSchemaName(), taskRecord.getTableName());
+ try (PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class)) {
+ boolean manual = false;
+ String reason = null;
+ if (!Strings.isNullOrEmpty(taskRecord.getData())) {
+ JsonNode data = JacksonUtil.getObjectReader().readTree(taskRecord.getData());
+ manual = data.path("manual").asBoolean(false);
+ reason = data.path("reason").asText(null);
+ }
+ Outcome outcome = VectorIndexRebuilder.rebuild(conn, indexName, manual,
+ manual && Strings.isNullOrEmpty(reason) ? VectorIndexRebuilder.MANUAL_REASON : reason);
+ if (outcome == Outcome.IN_PROGRESS) {
+ return new TaskResult(TaskResultCode.SKIPPED, "Vector index " + indexName + " is busy");
+ }
+ return new TaskResult(TaskResultCode.SUCCESS, outcome.name());
+ } catch (Throwable t) {
+ LOGGER.error("Rebuild of vector index {} failed", indexName, t);
+ return new TaskResult(TaskResultCode.FAIL, t.toString());
+ }
+ }
+
+ /** Retries rebuild execution on subsequent task sweeps if previously busy. */
+ @Override
+ public TaskResult checkCurrentResult(Task.TaskRecord taskRecord) {
+ return run(taskRecord);
+ }
+}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/VectorScorecardReconcileTask.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/VectorScorecardReconcileTask.java
new file mode 100644
index 0000000000..7494219f9d
--- /dev/null
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/VectorScorecardReconcileTask.java
@@ -0,0 +1,60 @@
+/*
+ * 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.coprocessor.tasks;
+
+import org.apache.phoenix.coprocessor.TaskRegionObserver.TaskResult;
+import org.apache.phoenix.coprocessor.TaskRegionObserver.TaskResultCode;
+import org.apache.phoenix.index.vector.VectorIndexRebuilder;
+import org.apache.phoenix.index.vector.VectorIndexRebuilder.ReconcileOutcome;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.SchemaUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Periodic server side task managing vector index drift scorecard reconciliation. Evaluates
+ * reconciliation intervals and triggers background rebuilds upon detecting drift.
+ */
+public class VectorScorecardReconcileTask extends BaseTask {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(VectorScorecardReconcileTask.class);
+
+ @Override
+ public TaskResult run(Task.TaskRecord taskRecord) {
+ String indexName =
+ SchemaUtil.getTableName(taskRecord.getSchemaName(), taskRecord.getTableName());
+ try (PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class)) {
+ if (VectorIndexRebuilder.reconcile(conn, indexName) == ReconcileOutcome.INDEX_DROPPED) {
+ return new TaskResult(TaskResultCode.SUCCESS, "Vector index " + indexName + " dropped");
+ }
+ } catch (Throwable t) {
+ LOGGER.warn("Reconciliation of vector index {} failed; the next sweep retries it", indexName,
+ t);
+ }
+ return new TaskResult(TaskResultCode.SKIPPED, "Awaiting the next reconciliation");
+ }
+
+ /** Executes periodic reconciliation check on each task table sweep. */
+ @Override
+ public TaskResult checkCurrentResult(Task.TaskRecord taskRecord) {
+ return run(taskRecord);
+ }
+}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
index cc07912a39..0800b28764 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
@@ -117,6 +117,7 @@ import org.apache.phoenix.hbase.index.write.LazyParallelWriterIndexCommitter;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.index.PhoenixIndexBuilderHelper;
import org.apache.phoenix.index.PhoenixIndexMetaData;
+import org.apache.phoenix.index.vector.ScorecardAccumulator;
import org.apache.phoenix.jdbc.HAGroupStoreManager;
import org.apache.phoenix.query.KeyRange;
import org.apache.phoenix.query.QueryConstants;
@@ -356,6 +357,8 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
private Set<ImmutableBytesPtr> rowsToLock = new TreeSet<>();
// The current and next states of the data rows corresponding to the pending mutations
private HashMap<ImmutableBytesPtr, Pair<Put, Put>> dataRowStates;
+ // Pending vector index scorecard deltas for this batch
+ private Map<ScorecardAccumulator.Key, long[]> scorecardDeltas;
// The previous concurrent batch contexts
private HashMap<ImmutableBytesPtr, BatchMutateContext> lastConcurrentBatchContext = null;
// The latches of the threads waiting for this batch to complete
@@ -551,6 +554,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
this.builder = new IndexBuildManager(env);
VectorCentroidCache.setServerConfiguration(env.getConfiguration());
+ ScorecardAccumulator.getInstance(env.getConfiguration());
// Clone the config since it is shared
DelegateRegionCoprocessorEnvironment indexWriterEnv =
new DelegateRegionCoprocessorEnvironment(env, ConnectionType.INDEX_WRITER_CONNECTION);
@@ -1436,6 +1440,11 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
IndexMaintainer.DeleteType.ALL_VERSIONS, ts);
indexUpdates.put(hTableInterfaceReference, del);
}
+ if (indexMaintainer.isMigrating()) {
+ deleteOutgoingIndexRow(indexMaintainer, currentDataRowVG, rowKeyPtr, ts,
+ encodedRegionName, indexPut.getRow(), indexRowKeyForCurrentDataRow,
+ hTableInterfaceReference, indexUpdates);
+ }
}
} else if (currentDataRowState != null && indexMaintainer.hasIndexRow(currentDataRowState)) {
if (indexMaintainer.isCDCIndex()) {
@@ -1448,13 +1457,59 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
indexUpdates.put(hTableInterfaceReference, getDeleteIndexMutation(cdcDataRowState,
indexMaintainer, ts, rowKeyPtr, encodedRegionName));
} else {
- indexUpdates.put(hTableInterfaceReference, getDeleteIndexMutation(currentDataRowState,
- indexMaintainer, ts, rowKeyPtr, encodedRegionName));
+ Mutation del = getDeleteIndexMutation(currentDataRowState, indexMaintainer, ts, rowKeyPtr,
+ encodedRegionName);
+ indexUpdates.put(hTableInterfaceReference, del);
+ if (indexMaintainer.isMigrating()) {
+ deleteOutgoingIndexRow(indexMaintainer,
+ new IndexUtil.SimpleValueGetter(currentDataRowState), rowKeyPtr, ts,
+ encodedRegionName, null, del.getRow(), hTableInterfaceReference, indexUpdates);
+ }
}
}
}
}
+ /**
+ * Deletes the index row corresponding to the outgoing generation during migration rebuilds.
+ */
+ private static void deleteOutgoingIndexRow(IndexMaintainer indexMaintainer,
+ ValueGetter currentDataRowVG, ImmutableBytesPtr rowKeyPtr, long ts, byte[] encodedRegionName,
+ byte[] writtenRowKey, byte[] deletedRowKey, HTableInterfaceReference hTableInterfaceReference,
+ ListMultimap<HTableInterfaceReference, Mutation> indexUpdates) {
+ byte[] outgoingRowKey =
+ indexMaintainer.buildOutgoingRowKey(currentDataRowVG, rowKeyPtr, ts, encodedRegionName);
+ if (
+ outgoingRowKey != null && !Bytes.equals(outgoingRowKey, writtenRowKey)
+ && !Bytes.equals(outgoingRowKey, deletedRowKey)
+ ) {
+ indexUpdates.put(hTableInterfaceReference, indexMaintainer
+ .buildRowDeleteMutation(outgoingRowKey, IndexMaintainer.DeleteType.ALL_VERSIONS, ts));
+ }
+ }
+
+ /**
+ * Derives scorecard deltas from index mutations for mutable vector indexes outside active
+ * migration.
+ */
+ private static void collectScorecardDeltas(BatchMutateContext context,
+ List<Pair<IndexMaintainer, HTableInterfaceReference>> indexTables, Put currentDataRowState,
+ ListMultimap<HTableInterfaceReference, Mutation> idxUpdates) {
+ if (context.immutableRows) {
+ return;
+ }
+ for (Pair<IndexMaintainer, HTableInterfaceReference> pair : indexTables) {
+ IndexMaintainer indexMaintainer = pair.getFirst();
+ if (indexMaintainer.isVectorIndex() && !indexMaintainer.isMigrating()) {
+ if (context.scorecardDeltas == null) {
+ context.scorecardDeltas = new HashMap<>();
+ }
+ ScorecardAccumulator.collect(indexMaintainer, currentDataRowState,
+ idxUpdates.get(pair.getSecond()), context.scorecardDeltas);
+ }
+ }
+ }
+
/**
* Generate the index update for a data row from the mutation that are obtained by merging the
* previous data row state with the pending row mutation.
@@ -1488,6 +1543,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
ListMultimap<HTableInterfaceReference, Mutation> idxUpdates = ArrayListMultimap.create();
generateIndexMutationsForRow(rowKeyPtr, currentDataRowState, nextDataRowState, ts,
encodedRegionName, QueryConstants.UNVERIFIED_BYTES, indexTables, idxUpdates);
+ collectScorecardDeltas(context, indexTables, currentDataRowState, idxUpdates);
for (Map.Entry<HTableInterfaceReference, Mutation> idxUpdate : idxUpdates.entries()) {
context.indexUpdates.put(idxUpdate.getKey(),
new Pair<>(idxUpdate.getValue(), rowKeyPtr.get()));
@@ -2233,6 +2289,10 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
if (success) { // The pre-index and data table updates are successful, and now, do post index
// updates
+ if (context.scorecardDeltas != null) {
+ ScorecardAccumulator.getInstance(c.getEnvironment().getConfiguration())
+ .accumulate(context.scorecardDeltas);
+ }
doPost(c, context);
}
} finally {
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/index/vector/ScorecardAccumulator.java b/phoenix-core-server/src/main/java/org/apache/phoenix/index/vector/ScorecardAccumulator.java
new file mode 100644
index 0000000000..561947451e
--- /dev/null
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/index/vector/ScorecardAccumulator.java
@@ -0,0 +1,281 @@
+/*
+ * 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.GENERATION_ID;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_NAME;
+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.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+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 java.util.stream.Collectors;
+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;
+ }
+ merge(entry.getKey(), delta.clone());
+ if (delta[0] > 0) {
+ metrics.incrementVectorCentroidAssignments(entry.getKey().indexName, delta[0]);
+ }
+ if (delta[1] > 0) {
+ metrics.incrementVectorCentroidReassignments(entry.getKey().indexName, delta[1]);
+ }
+ }
+ }
+
+ private void merge(Key key, long[] delta) {
+ deltas.merge(key, delta, (a, b) -> new long[] { a[0] + b[0], a[1] + b[1] });
+ }
+
+ private void flushQuietly() {
+ try {
+ flush();
+ } catch (Throwable t) {
+ LOGGER.warn("Vector index scorecard flush failed; deltas are retained for the next flush", t);
+ }
+ }
+
+ /**
+ * Flushes buffered deltas to {@code SYSTEM.VECTOR_CENTROID} using atomic counter increments, one
+ * commit per index generation. Deltas for dropped indexes or retired generations are discarded.
+ * Deltas of a generation whose commit did not complete are returned to the buffer and retried by
+ * the next flush, since reassignment counts, unlike populations, cannot be recovered by
+ * reconciliation. A commit that fails after partially applying may therefore be counted twice.
+ */
+ public synchronized void flush() throws SQLException {
+ if (deltas.isEmpty()) {
+ return;
+ }
+ Map<Key, Map<Key, long[]>> byGeneration = new HashMap<>();
+ for (Key key : deltas.keySet()) {
+ long[] delta = deltas.remove(key);
+ if (delta != null) {
+ byGeneration.computeIfAbsent(key.generationKey(), k -> new HashMap<>()).put(key, delta);
+ }
+ }
+ List<String> indexNames =
+ byGeneration.keySet().stream().map(k -> k.indexName).distinct().collect(Collectors.toList());
+ long start = EnvironmentEdgeManager.currentTimeMillis();
+ try (Connection conn = QueryUtil.getConnectionOnServer(conf)) {
+ Iterator<Map.Entry<Key, Map<Key, long[]>>> pending = byGeneration.entrySet().iterator();
+ while (pending.hasNext()) {
+ Map.Entry<Key, Map<Key, long[]>> entry = pending.next();
+ Key generation = entry.getKey();
+ if (isLive(conn, generation)) {
+ List<ScorecardRow> rows = new ArrayList<>(entry.getValue().size());
+ for (Map.Entry<Key, long[]> delta : entry.getValue().entrySet()) {
+ rows.add(new ScorecardRow(delta.getKey().centroidId, delta.getValue()[0],
+ delta.getValue()[1]));
+ }
+ CentroidManager.adjustScorecard(conn, generation.indexName, generation.generation, rows);
+ }
+ pending.remove();
+ }
+ } finally {
+ for (Map<Key, long[]> unflushed : byGeneration.values()) {
+ unflushed.forEach(this::merge);
+ }
+ }
+ long elapsed = EnvironmentEdgeManager.currentTimeMillis() - start;
+ MetricsVectorIndexSource metrics =
+ MetricsIndexerSourceFactory.getInstance().getMetricsVectorIndexSource();
+ indexNames.forEach(name -> metrics.updateVectorScorecardFlushTime(name, elapsed));
+ }
+
+ /**
+ * Verifies whether centroid metadata for the specified generation remains active. A failed read
+ * propagates, so that a catalog outage retains deltas rather than treating them as retired.
+ */
+ private static boolean isLive(Connection conn, Key generationKey) throws SQLException {
+ try (PreparedStatement ps = conn.prepareStatement("SELECT 1 FROM " + SYSTEM_VECTOR_CENTROID_NAME
+ + " WHERE " + INDEX_NAME + " = ? AND " + GENERATION_ID + " = ? AND " + CENTROID_ID
+ + " >= 0 AND " + CENTROID_VECTOR + " IS NOT NULL LIMIT 1")) {
+ ps.setString(1, generationKey.indexName);
+ ps.setLong(2, generationKey.generation);
+ try (ResultSet rs = ps.executeQuery()) {
+ return rs.next();
+ }
+ }
+ }
+
+ /** Composite key identifying a specific centroid within a generation and index. */
+ public static final class Key {
+ private final String indexName;
+ private final long generation;
+ private final int centroidId;
+
+ public Key(String indexName, long generation, int centroidId) {
+ this.indexName = Objects.requireNonNull(indexName, "indexName");
+ this.generation = generation;
+ this.centroidId = centroidId;
+ }
+
+ Key generationKey() {
+ return new Key(indexName, generation, 0);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof Key)) {
+ return false;
+ }
+ Key other = (Key) o;
+ return generation == other.generation && centroidId == other.centroidId
+ && indexName.equals(other.indexName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(indexName, generation, centroidId);
+ }
+
+ @Override
+ public String toString() {
+ return indexName + ":" + generation + ":" + centroidId;
+ }
+ }
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorCentroidTableIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorCentroidTableIT.java
index c5688f12aa..317ca55a3f 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorCentroidTableIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorCentroidTableIT.java
@@ -30,6 +30,7 @@ import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.sql.Connection;
@@ -43,6 +44,9 @@ import java.sql.Types;
import java.util.Arrays;
import java.util.List;
import org.apache.phoenix.index.vector.CentroidManager;
+import org.apache.phoenix.index.vector.ClusterSkewMetrics;
+import org.apache.phoenix.index.vector.GenerationSummary;
+import org.apache.phoenix.index.vector.ScorecardRow;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PTable;
@@ -66,7 +70,8 @@ public class VectorCentroidTableIT extends ParallelStatsDisabledIT {
try (Statement stmt = conn.createStatement(); ResultSet rs =
stmt.executeQuery("SELECT * FROM " + SYSTEM_VECTOR_CENTROID_NAME + " WHERE 1=0")) {
ResultSetMetaData rsmd = rs.getMetaData();
- assertEquals(4, rsmd.getColumnCount());
+ // Key, centroid vector, scorecard metrics, and generation summary columns
+ assertEquals(12, rsmd.getColumnCount());
assertEquals(INDEX_NAME, rsmd.getColumnName(1));
assertEquals(Types.VARCHAR, rsmd.getColumnType(1));
@@ -211,6 +216,17 @@ public class VectorCentroidTableIT extends ParallelStatsDisabledIT {
assertEquals(CENTROID_VECTOR, rs.getString(COLUMN_NAME));
assertEquals(Types.VARBINARY, rs.getInt(DATA_TYPE));
+ // Per-centroid scorecard and per-generation summary columns
+ String[] names = { "CLUSTER_SIZE", "REASSIGN_COUNT", "SKEW_METRICS", "REBUILD_STATE",
+ "TRIGGER_REASON", "REQUESTED_LISTS", "LAST_REBUILD_TIME", "LAST_SCORECARD_UPDATE" };
+ int[] types = { Types.BIGINT, Types.BIGINT, Types.VARBINARY, Types.CHAR, Types.VARCHAR,
+ Types.INTEGER, Types.BIGINT, Types.BIGINT };
+ for (int i = 0; i < names.length; i++) {
+ assertTrue("Expected " + names[i] + " column", rs.next());
+ assertEquals(names[i], rs.getString(COLUMN_NAME));
+ assertEquals(types[i], rs.getInt(DATA_TYPE));
+ }
+
assertFalse("No additional columns expected", rs.next());
}
}
@@ -313,4 +329,80 @@ public class VectorCentroidTableIT extends ParallelStatsDisabledIT {
after.getLastDDLTimestamp() > before.getLastDDLTimestamp());
}
}
+
+ /** Verifies generation summary persistence and partial field updates. */
+ @Test
+ public void testGenerationSummaryRoundTrip() throws Exception {
+ String indexName = "TEST_VECTOR_IDX_" + generateUniqueName();
+ ClusterSkewMetrics skew = ClusterSkewMetrics.compute(new int[] { 10, 20, 30 });
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ assertNull(CentroidManager.loadGenerationSummary(conn, indexName, 5L));
+ CentroidManager.persistGenerationSummary(conn, indexName, 5L,
+ new GenerationSummary(GenerationSummary.BUILDING, "SKEW", 4, skew, null, null));
+ CentroidManager.persistGenerationSummary(conn, indexName, 5L,
+ new GenerationSummary(GenerationSummary.ACTIVE, null, null, null, 123L, 456L));
+ GenerationSummary summary = CentroidManager.loadGenerationSummary(conn, indexName, 5L);
+ assertEquals(GenerationSummary.ACTIVE, summary.getRebuildState());
+ assertEquals("SKEW", summary.getTriggerReason());
+ assertEquals(Integer.valueOf(4), summary.getRequestedLists());
+ assertEquals(skew, summary.getSkewMetrics());
+ assertEquals(Long.valueOf(123L), summary.getLastRebuildTime());
+ assertEquals(Long.valueOf(456L), summary.getLastScorecardUpdate());
+ assertNull(CentroidManager.loadGenerationSummary(conn, indexName, 6L));
+ }
+ }
+
+ /**
+ * Verifies that scorecard metrics are scoped to individual generations and cleaned up upon
+ * retirement.
+ */
+ @Test
+ public void testScorecardIsGenerationScoped() throws Exception {
+ String indexName = "TEST_VECTOR_IDX_" + generateUniqueName();
+ List<float[]> centroids = Arrays.asList(new float[] { 1 }, new float[] { 2 });
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ CentroidManager.persistCentroids(conn, indexName, 1L, centroids);
+ CentroidManager.persistCentroids(conn, indexName, 2L, centroids);
+ CentroidManager.persistGenerationSummary(conn, indexName, 1L,
+ new GenerationSummary(GenerationSummary.ACTIVE, null, 2, null, null, 1L));
+ CentroidManager.adjustScorecard(conn, indexName, 1L,
+ Arrays.asList(new ScorecardRow(0, 7, 1), new ScorecardRow(1, 3, 0)));
+
+ List<ScorecardRow> gen1 = CentroidManager.loadScorecard(conn, indexName, 1L);
+ assertEquals(2, gen1.size());
+ assertEquals(7, gen1.get(0).getClusterSize());
+ assertEquals(1, gen1.get(0).getReassignCount());
+ assertEquals(3, gen1.get(1).getClusterSize());
+ List<ScorecardRow> gen2 = CentroidManager.loadScorecard(conn, indexName, 2L);
+ assertEquals(0, gen2.get(0).getClusterSize());
+ assertEquals(2, CentroidManager.loadCentroids(conn, indexName, 1L).size());
+ assertEquals(Arrays.asList(1L, 2L), CentroidManager.listGenerations(conn, indexName));
+
+ CentroidManager.deleteGeneration(conn, indexName, 1L);
+ assertTrue(CentroidManager.loadScorecard(conn, indexName, 1L).isEmpty());
+ assertNull(CentroidManager.loadGenerationSummary(conn, indexName, 1L));
+ assertEquals(2, CentroidManager.loadScorecard(conn, indexName, 2L).size());
+ assertEquals(Arrays.asList(2L), CentroidManager.listGenerations(conn, indexName));
+ }
+ }
+
+ /** Verifies rebuild lock exclusivity and expiration semantics. */
+ @Test
+ public void testRebuildClaim() throws Exception {
+ String indexName = "TEST_VECTOR_IDX_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ assertTrue(CentroidManager.claimRebuild(conn, indexName, "a", 60000));
+ assertFalse(CentroidManager.claimRebuild(conn, indexName, "b", 60000));
+ assertTrue(CentroidManager.listGenerations(conn, indexName).isEmpty());
+ CentroidManager.releaseRebuild(conn, indexName, "b");
+ assertFalse("Only the holder releases",
+ CentroidManager.claimRebuild(conn, indexName, "b", 60000));
+ CentroidManager.releaseRebuild(conn, indexName, "a");
+ assertTrue(CentroidManager.claimRebuild(conn, indexName, "b", 60000));
+ Thread.sleep(5);
+ assertTrue("An expired claim is taken over",
+ CentroidManager.claimRebuild(conn, indexName, "c", 1));
+ assertFalse(CentroidManager.claimRebuild(conn, indexName, "b", 60000));
+ }
+ }
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexRebuildIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexRebuildIT.java
new file mode 100644
index 0000000000..e788978233
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexRebuildIT.java
@@ -0,0 +1,872 @@
+/*
+ * 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.apache.phoenix.end2end.VectorIndexTestUtil.KNOWN_CENTROIDS;
+import static org.apache.phoenix.end2end.VectorIndexTestUtil.assertIndexConsistent;
+import static org.apache.phoenix.end2end.VectorIndexTestUtil.assertIndexVerifies;
+import static org.apache.phoenix.end2end.VectorIndexTestUtil.bruteForceTopK;
+import static org.apache.phoenix.end2end.VectorIndexTestUtil.countCentroids;
+import static org.apache.phoenix.end2end.VectorIndexTestUtil.setupTableAndKnownCentroids;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_TASK_HBASE_TABLE_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_VECTOR_CENTROID_HBASE_TABLE_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_VECTOR_CENTROID_NAME;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_TASK_HANDLING_MAX_INTERVAL_MS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.sql.Array;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.index.vector.CentroidManager;
+import org.apache.phoenix.index.vector.GenerationSummary;
+import org.apache.phoenix.index.vector.ScorecardAccumulator;
+import org.apache.phoenix.index.vector.ScorecardRow;
+import org.apache.phoenix.index.vector.VectorIndexRebuilder;
+import org.apache.phoenix.index.vector.VectorIndexRebuilder.Outcome;
+import org.apache.phoenix.index.vector.VectorIndexRebuilderTestHooks;
+import org.apache.phoenix.index.vector.VectorIndexScorecard;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.JacksonUtil;
+import org.bson.BinaryVector;
+import org.bson.BsonBinary;
+import org.bson.BsonDocument;
+import org.junit.After;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/**
+ * Integration tests for vector index drift scorecarding, periodic reconciliation, and generational
+ * online rebuild migration.
+ */
+@Category(ParallelStatsDisabledTest.class)
+public class VectorIndexRebuildIT extends ParallelStatsDisabledIT {
+
+ private static RegionCoprocessorEnvironment taskRegionEnvironment;
+
+ @BeforeClass
+ public static synchronized void findTaskRegion() throws Exception {
+ taskRegionEnvironment = (RegionCoprocessorEnvironment) getUtility()
+ .getRSForFirstRegionInTable(SYSTEM_TASK_HBASE_TABLE_NAME)
+ .getRegions(SYSTEM_TASK_HBASE_TABLE_NAME).get(0).getCoprocessorHost()
+ .findCoprocessorEnvironment(TaskRegionObserver.class.getName());
+ }
+
+ @After
+ public void clearHook() {
+ VectorIndexRebuilderTestHooks.setMigrationHook(null);
+ }
+
+ /**
+ * Returns test connection properties with zero intervals for metadata refresh, reconciliation,
+ * and rebuild gating.
+ */
+ private static Properties props(boolean autoRebuild) {
+ Properties props = new Properties();
+ props.setProperty(QueryServices.INDEX_POPULATION_SLEEP_TIME, "0");
+ props.setProperty(QueryServices.VECTOR_SCORECARD_RECONCILE_INTERVAL_MS_ATTRIB, "0");
+ props.setProperty(QueryServices.VECTOR_DRIFT_MIN_POPULATION_ATTRIB, "10");
+ props.setProperty(QueryServices.VECTOR_REBUILD_AUTO_ENABLED_ATTRIB,
+ Boolean.toString(autoRebuild));
+ props.setProperty(QueryServices.VECTOR_REBUILD_MIN_INTERVAL_MS_ATTRIB, "0");
+ return props;
+ }
+
+ private static Connection connect(boolean autoRebuild) throws SQLException {
+ return DriverManager.getConnection(getUrl(), props(autoRebuild));
+ }
+
+ /**
+ * Returns a vector offset from centroid {@code c} of {@link VectorIndexTestUtil#KNOWN_CENTROIDS}.
+ */
+ private static Float[] near(int c, float jitter) {
+ float[] centroid = KNOWN_CENTROIDS.get(c);
+ Float[] v = new Float[4];
+ for (int i = 0; i < 4; i++) {
+ v[i] = centroid[i] + jitter;
+ }
+ return v;
+ }
+
+ private static void upsert(Connection conn, String table, String id, Float[] v, String label)
+ throws SQLException {
+ try (PreparedStatement ps =
+ conn.prepareStatement("UPSERT INTO " + table + " (ID, V, LABEL) VALUES (?, ?, ?)")) {
+ ps.setString(1, id);
+ Array array = conn.createArrayOf("FLOAT", v);
+ ps.setArray(2, array);
+ ps.setString(3, label);
+ ps.executeUpdate();
+ }
+ }
+
+ private static void flushScorecards() throws SQLException {
+ ScorecardAccumulator.getInstance().flush();
+ }
+
+ private static long[] scorecard(Connection conn, String index, long generation)
+ throws SQLException {
+ // In centroid id order
+ List<ScorecardRow> rows = CentroidManager.loadScorecard(conn, index, generation);
+ long[] sizes = new long[rows.size()];
+ for (int i = 0; i < sizes.length; i++) {
+ sizes[i] = rows.get(i).getClusterSize();
+ }
+ return sizes;
+ }
+
+ private static long[] groupedCount(Connection conn, String index, int lists) throws SQLException {
+ long[] counts = new long[lists];
+ try (ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT \"_CENTROID_ID\", COUNT(*) FROM " + index + " GROUP BY \"_CENTROID_ID\"")) {
+ while (rs.next()) {
+ counts[rs.getInt(1)] = rs.getLong(2);
+ }
+ }
+ return counts;
+ }
+
+ private static long count(Connection conn, String table) throws SQLException {
+ try (ResultSet rs = conn.createStatement().executeQuery("SELECT COUNT(*) FROM " + table)) {
+ assertTrue(rs.next());
+ return rs.getLong(1);
+ }
+ }
+
+ private static PTable index(Connection conn, String index) throws SQLException {
+ return conn.unwrap(PhoenixConnection.class).getTableNoCache(index);
+ }
+
+ private static long reassigned(Connection conn, String index, long generation, int centroid)
+ throws SQLException {
+ return CentroidManager.loadScorecard(conn, index, generation).get(centroid).getReassignCount();
+ }
+
+ /**
+ * Returns the reasons carried by the automatic rebuild tasks queued for an index, as the rebuild
+ * task reads them.
+ */
+ private static List<String> queuedAutomaticRebuilds(Connection conn, String index)
+ throws Exception {
+ List<String> reasons = new ArrayList<>();
+ for (Task.TaskRecord task : Task.queryTaskTable(conn, null, null, index,
+ PTable.TaskType.VECTOR_INDEX_REBUILD, null, null)) {
+ JsonNode data = JacksonUtil.getObjectReader().readTree(task.getData());
+ if (!data.path("manual").asBoolean(false)) {
+ reasons.add(data.path("reason").asText(null));
+ }
+ }
+ return reasons;
+ }
+
+ private static void runTaskSweep() {
+ new TaskRegionObserver.SelfHealingTask(taskRegionEnvironment,
+ DEFAULT_TASK_HANDLING_MAX_INTERVAL_MS).run();
+ }
+
+ /**
+ * Verifies that inline scorecard maintenance tracks inserts, deletes, cross-centroid updates, and
+ * covered column updates matching grouped index counts.
+ */
+ @Test
+ public void testInlineCountsMatchGroupedCount() throws Exception {
+ String table = "T_SC_" + generateUniqueName();
+ String index = "IDX_SC_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long gen = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 12; i++) {
+ upsert(conn, table, "r" + i, near(i % 4, 0.01f), "l" + i);
+ }
+ conn.commit();
+ upsert(conn, table, "r0", near(1, 0.02f), "l0"); // centroid 0 -> 1
+ upsert(conn, table, "r1", near(1, 0.03f), "l1"); // within centroid 1
+ upsert(conn, table, "r2", near(2, 0.01f), "changed"); // covered column only
+ conn.commit();
+ conn.createStatement().execute("DELETE FROM " + table + " WHERE ID IN ('r3', 'r7')");
+ conn.commit();
+ flushScorecards();
+
+ long[] expected = groupedCount(conn, index, 4);
+ assertEquals(2, expected[0]);
+ assertEquals(4, expected[1]);
+ assertEquals(3, expected[2]);
+ assertEquals(1, expected[3]);
+ assertEquals(java.util.Arrays.toString(expected),
+ java.util.Arrays.toString(scorecard(conn, index, gen)));
+ assertEquals("The move into centroid 1 counts a reassignment", 1,
+ reassigned(conn, index, gen, 1));
+ assertEquals(0, reassigned(conn, index, gen, 0));
+ }
+ }
+
+ /** Verifies that replaying an already applied batch produces no net delta in scorecard counts. */
+ @Test
+ public void testReplayIsNeutral() throws Exception {
+ String table = "T_RP_" + generateUniqueName();
+ String index = "IDX_RP_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long gen = index(conn, index).getVectorCentroidGeneration();
+ for (int round = 0; round < 2; round++) {
+ for (int i = 0; i < 8; i++) {
+ upsert(conn, table, "r" + i, near(i % 4, 0.01f), "l" + i);
+ }
+ conn.commit();
+ flushScorecards();
+ assertEquals("[2, 2, 2, 2]", java.util.Arrays.toString(scorecard(conn, index, gen)));
+ }
+ }
+ }
+
+ /** Verifies that scorecard flushes incrementally add to stored counts. */
+ @Test
+ public void testFlushesAccumulate() throws Exception {
+ String index = "IDX_FL_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ CentroidManager.persistCentroids(conn, index, 1L,
+ java.util.Arrays.asList(new float[] { 1 }, new float[] { 2 }));
+ for (long delta : new long[] { 5, 7 }) {
+ Map<ScorecardAccumulator.Key, long[]> batch = new HashMap<>();
+ batch.put(new ScorecardAccumulator.Key(index, 1L, 0), new long[] { delta, 0 });
+ ScorecardAccumulator.getInstance().accumulate(batch);
+ flushScorecards();
+ }
+ assertEquals(12, scorecard(conn, index, 1L)[0]);
+ }
+ }
+
+ /** Verifies that buffered scorecard deltas for retired generations are discarded during flush. */
+ @Test
+ public void testFlushDoesNotResurrectRetiredGeneration() throws Exception {
+ String index = "IDX_RT_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ CentroidManager.persistCentroids(conn, index, 1L, java.util.Arrays.asList(new float[] { 1 }));
+ Map<ScorecardAccumulator.Key, long[]> batch = new HashMap<>();
+ batch.put(new ScorecardAccumulator.Key(index, 1L, 0), new long[] { 3, 0 });
+ ScorecardAccumulator.getInstance().accumulate(batch);
+ CentroidManager.deleteGeneration(conn, index, 1L);
+ flushScorecards();
+ assertTrue(CentroidManager.listGenerations(conn, index).isEmpty());
+ try (ResultSet rs = conn.createStatement().executeQuery("SELECT COUNT(*) FROM "
+ + SYSTEM_VECTOR_CENTROID_NAME + " WHERE INDEX_NAME = '" + index + "'")) {
+ assertTrue(rs.next());
+ assertEquals(0, rs.getInt(1));
+ }
+ }
+ }
+
+ /**
+ * Verifies that index creation initializes scorecard counts and enqueues the initial
+ * reconciliation task.
+ */
+ @Test
+ public void testScorecardSeededAtBuildAndTaskEnqueued() throws Exception {
+ String table = "T_SD_" + generateUniqueName();
+ String index = "IDX_SD_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ conn.createStatement().execute("CREATE TABLE " + table
+ + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 4), LABEL VARCHAR)");
+ VectorIndexTestUtil.loadClusteredVectors(conn, table, 100);
+ conn.createStatement().execute("CREATE VECTOR INDEX " + index + " ON " + table
+ + " (V) INCLUDE (LABEL) WITH (algorithm='IVF', metric='L2', lists=4, sample_size=100)");
+ PTable idx = index(conn, index);
+ long total = 0;
+ for (long size : scorecard(conn, index, idx.getVectorCentroidGeneration())) {
+ total += size;
+ }
+ assertEquals(100, total);
+ GenerationSummary summary =
+ CentroidManager.loadGenerationSummary(conn, index, idx.getVectorCentroidGeneration());
+ assertEquals(GenerationSummary.ACTIVE, summary.getRebuildState());
+ assertEquals(Integer.valueOf(4), summary.getRequestedLists());
+ assertNotNull(summary.getLastScorecardUpdate());
+ List<Task.TaskRecord> tasks = Task.queryTaskTable(conn, null, null, index,
+ PTable.TaskType.VECTOR_SCORECARD_RECONCILE, null, null);
+ assertEquals(1, tasks.size());
+ }
+ }
+
+ /**
+ * Verifies that scorecard reconciliation recalculates posting list populations, clears
+ * reassignment counts, and updates timestamp metadata.
+ */
+ @Test
+ public void testReconcileRepairsDivergence() throws Exception {
+ String table = "T_RC_" + generateUniqueName();
+ String index = "IDX_RC_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long gen = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 6; i++) {
+ upsert(conn, table, "r" + i, near(i % 3, 0.01f), "l");
+ }
+ conn.commit();
+ flushScorecards();
+ // Corrupt the counters: centroids 0 and 3 hold 2 and 0 rows
+ CentroidManager.adjustScorecard(conn, index, gen,
+ java.util.Arrays.asList(new ScorecardRow(0, 97, 5), new ScorecardRow(3, 42, 0)));
+ assertEquals(VectorIndexRebuilder.ReconcileOutcome.RECONCILED,
+ VectorIndexRebuilder.reconcile(conn.unwrap(PhoenixConnection.class), index));
+ assertEquals("[2, 2, 2, 0]", java.util.Arrays.toString(scorecard(conn, index, gen)));
+ assertEquals(0, reassigned(conn, index, gen, 0));
+ assertNotNull(
+ CentroidManager.loadGenerationSummary(conn, index, gen).getLastScorecardUpdate());
+ }
+ }
+
+ /**
+ * Verifies that deltas a failed flush could not commit are kept and written by the next flush, so
+ * reassignments, which reconciliation cannot recover, survive a catalog outage.
+ */
+ @Test
+ public void testFailedFlushRetainsDeltas() throws Exception {
+ String table = "T_FF_" + generateUniqueName();
+ String index = "IDX_FF_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long gen = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 4; i++) {
+ upsert(conn, table, "r" + i, near(i, 0.01f), "l");
+ }
+ conn.commit();
+ flushScorecards();
+ upsert(conn, table, "r0", near(1, 0.02f), "l"); // centroid 0 -> 1
+ conn.commit();
+ try (Admin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin()) {
+ admin.disableTable(SYSTEM_VECTOR_CENTROID_HBASE_TABLE_NAME);
+ try {
+ flushScorecards();
+ fail("Flush to a disabled scorecard table should fail");
+ } catch (SQLException expected) {
+ // The deltas are retained
+ } finally {
+ admin.enableTable(SYSTEM_VECTOR_CENTROID_HBASE_TABLE_NAME);
+ }
+ }
+ flushScorecards();
+ assertEquals("[0, 2, 1, 1]", java.util.Arrays.toString(scorecard(conn, index, gen)));
+ assertEquals(1, reassigned(conn, index, gen, 1));
+ }
+ }
+
+ /**
+ * Verifies that a RegionServer flush committed after reconciliation reads the scorecard, and
+ * before it writes, survives: reconciliation corrects relative to what it read rather than
+ * overwriting the counters with absolute values.
+ */
+ @Test
+ public void testReconcilePreservesConcurrentFlush() throws Exception {
+ String table = "T_RF_" + generateUniqueName();
+ String index = "IDX_RF_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long gen = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 6; i++) {
+ upsert(conn, table, "r" + i, near(i % 3, 0.01f), "l");
+ }
+ conn.commit();
+ flushScorecards();
+ CentroidManager.adjustScorecard(conn, index, gen,
+ java.util.Arrays.asList(new ScorecardRow(0, 7, 4)));
+ VectorIndexRebuilderTestHooks.setReconcileHook((name, generation) -> {
+ // A flush of one insert and three reassignments into centroid 0
+ try (Connection c = connect(false)) {
+ CentroidManager.adjustScorecard(c, name, generation,
+ java.util.Arrays.asList(new ScorecardRow(0, 1, 3)));
+ }
+ });
+ try {
+ VectorIndexScorecard.reconcile(conn.unwrap(PhoenixConnection.class), index(conn, index),
+ gen);
+ } finally {
+ VectorIndexRebuilderTestHooks.setReconcileHook(null);
+ }
+ // Centroid 0 holds 2 rows; the corrupt +7 and the 4 prior reassignments are corrected away,
+ // and the concurrent flush's +1 and 3 reassignments are kept
+ assertEquals("[3, 2, 2, 0]", java.util.Arrays.toString(scorecard(conn, index, gen)));
+ assertEquals(3, reassigned(conn, index, gen, 0));
+ }
+ }
+
+ /**
+ * Verifies that reconciliation tasks execute recurringly and clean up upon index drop.
+ */
+ @Test
+ public void testReconcileTaskRecursAndRetiresOnDrop() throws Exception {
+ String table = "T_TK_" + generateUniqueName();
+ String index = "IDX_TK_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long gen = index(conn, index).getVectorCentroidGeneration();
+ upsert(conn, table, "a", near(0, 0.01f), "l");
+ conn.commit();
+ flushScorecards();
+ for (int sweep = 0; sweep < 2; sweep++) {
+ // Corrupt centroid 0, which holds 1 row, to 50
+ CentroidManager.adjustScorecard(conn, index, gen,
+ java.util.Arrays.asList(new ScorecardRow(0, 49, 0)));
+ // Reset timestamp to force task execution on next sweep
+ conn.createStatement()
+ .execute("UPSERT INTO " + SYSTEM_VECTOR_CENTROID_NAME
+ + " (INDEX_NAME, GENERATION_ID, CENTROID_ID, LAST_SCORECARD_UPDATE) VALUES ('" + index
+ + "', " + gen + ", -1, NULL)");
+ conn.commit();
+ runTaskSweep();
+ assertEquals("sweep " + sweep, 1, scorecard(conn, index, gen)[0]);
+ Task.TaskRecord task = Task.queryTaskTable(conn, null, null, index,
+ PTable.TaskType.VECTOR_SCORECARD_RECONCILE, null, null).get(0);
+ assertEquals(PTable.TaskStatus.STARTED.toString(), task.getStatus());
+ }
+ conn.createStatement().execute("DROP INDEX " + index + " ON " + table);
+ assertTrue(Task.queryTaskTable(conn, null, null, index,
+ PTable.TaskType.VECTOR_SCORECARD_RECONCILE, null, null).isEmpty());
+ assertEquals(0, countCentroids(conn, index, null));
+ }
+ }
+
+ /**
+ * Verifies synchronous ALTER INDEX REBUILD execution, including model retraining, row migration,
+ * retirement of prior generation, and query result accuracy.
+ */
+ @Test
+ public void testManualRebuild() throws Exception {
+ String table = "T_MR_" + generateUniqueName();
+ String index = "IDX_MR_" + generateUniqueName();
+ Random rng = new Random(7);
+ Map<String, float[]> rows = new LinkedHashMap<>();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long before = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 200; i++) {
+ Float[] v = new Float[4];
+ float[] f = new float[4];
+ for (int d = 0; d < 4; d++) {
+ f[d] = rng.nextFloat() * 10;
+ v[d] = f[d];
+ }
+ upsert(conn, table, String.format("r%03d", i), v, "l" + i);
+ rows.put(String.format("r%03d", i), f);
+ }
+ conn.commit();
+ conn.createStatement().execute("ALTER INDEX " + index + " ON " + table + " REBUILD");
+
+ PTable idx = index(conn, index);
+ long after = idx.getVectorCentroidGeneration();
+ assertTrue(after > before);
+ assertFalse(idx.isVectorRebuildInProgress());
+ assertEquals(0, countCentroids(conn, index, before));
+ assertEquals(idx.getVectorIvfLists().intValue(), countCentroids(conn, index, after));
+ assertEquals(200, count(conn, index));
+ assertIndexVerifies(table, index, 200);
+ long total = 0;
+ for (long size : scorecard(conn, index, after)) {
+ total += size;
+ }
+ assertEquals(200, total);
+ GenerationSummary summary = CentroidManager.loadGenerationSummary(conn, index, after);
+ assertEquals(GenerationSummary.ACTIVE, summary.getRebuildState());
+ assertEquals(VectorIndexRebuilder.MANUAL_REASON, summary.getTriggerReason());
+ assertNotNull(summary.getLastRebuildTime());
+
+ float[] q = new float[] { 5, 5, 5, 5 };
+ assertEquals(bruteForceTopK(rows, q, "L2", 5),
+ topK(conn, table, "/*+ VECTOR_PROBE_COUNT(" + idx.getVectorIvfLists() + ") */", q, 5));
+ }
+ }
+
+ /**
+ * Verifies dual-generation coexistence and write fencing during an active index rebuild
+ * migration, and that queries probe both generations so rows already migrated into the building
+ * generation's posting lists remain visible.
+ */
+ @Test
+ public void testMigrationIsFencedAndQueriesProbeBothGenerations() throws Exception {
+ String table = "T_MG_" + generateUniqueName();
+ String index = "IDX_MG_" + generateUniqueName();
+ Map<String, float[]> rows = new LinkedHashMap<>();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long active = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 40; i++) {
+ Float[] v = near(i % 4, 0.01f * i);
+ upsert(conn, table, "r" + i, v, "l" + i);
+ rows.put("r" + i, unbox(v));
+ }
+ conn.commit();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ VectorIndexRebuilderTestHooks.setMigrationHook((idx, building) -> {
+ try (Connection c = connect(false)) {
+ PTable current = index(c, index);
+ assertEquals(Long.valueOf(active), current.getVectorCentroidGeneration());
+ assertEquals(Long.valueOf(building), current.getVectorBuildingGeneration());
+ assertTrue(countCentroids(c, index, active) > 0);
+ assertTrue(countCentroids(c, index, building) > 0);
+ assertEquals(count(c, table), count(c, index));
+
+ // Concurrent writes during active migration
+ Float[] moved = near(3, 0.5f);
+ upsert(c, table, "r0", moved, "moved");
+ upsert(c, table, "r1", near(1, 0.01f), "covered-only");
+ upsert(c, table, "new", near(2, 0.2f), "new");
+ c.createStatement().execute("DELETE FROM " + table + " WHERE ID = 'r2'");
+ c.commit();
+ assertEquals(count(c, table), count(c, index));
+
+ String explain = explain(c, table, new float[] { 1, 0, 0, 0 });
+ assertTrue(explain, explain.contains("ACROSS ACTIVE AND BUILDING GENERATIONS"));
+ // The build pass has moved every row under the building generation's centroid IDs, so
+ // a plan probing only the active generation finds nothing
+ Map<String, float[]> expected = new LinkedHashMap<>(rows);
+ expected.put("r0", unbox(moved));
+ expected.put("new", unbox(near(2, 0.2f)));
+ expected.remove("r2");
+ for (int k = 0; k < KNOWN_CENTROIDS.size(); k++) {
+ float[] q = unbox(near(k, 0.005f));
+ assertEquals("cluster " + k, bruteForceTopK(expected, q, "L2", 5),
+ topK(c, table, "", q, 5));
+ }
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ conn.createStatement().execute("ALTER INDEX " + index + " ON " + table + " REBUILD");
+ if (failure.get() != null) {
+ throw new AssertionError(failure.get());
+ }
+ rows.put("r0", unbox(near(3, 0.5f)));
+ rows.put("new", unbox(near(2, 0.2f)));
+ rows.remove("r2");
+ assertEquals(rows.size(), count(conn, index));
+ assertIndexConsistent(table, index);
+ PTable idx = index(conn, index);
+ assertEquals(0, countCentroids(conn, index, active));
+ float[] q = new float[] { 0, 1, 0, 0 };
+ assertEquals(bruteForceTopK(rows, q, "L2", 5),
+ topK(conn, table, "/*+ VECTOR_PROBE_COUNT(" + idx.getVectorIvfLists() + ") */", q, 5));
+ }
+ }
+
+ /**
+ * Verifies that a migration interrupted by failure preserves building generation state and can
+ * resume to completion.
+ */
+ @Test
+ public void testFailedMigrationResumes() throws Exception {
+ String table = "T_FM_" + generateUniqueName();
+ String index = "IDX_FM_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long active = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 40; i++) {
+ upsert(conn, table, "r" + i, near(i % 4, 0.01f * i), "l" + i);
+ }
+ conn.commit();
+ VectorIndexRebuilderTestHooks.setMigrationHook((idx, building) -> {
+ throw new RuntimeException("injected");
+ });
+ try {
+ conn.createStatement().execute("ALTER INDEX " + index + " ON " + table + " REBUILD");
+ fail();
+ } catch (SQLException e) {
+ // expected
+ }
+ VectorIndexRebuilderTestHooks.setMigrationHook(null);
+ PTable failed = index(conn, index);
+ assertTrue(failed.isVectorRebuildInProgress());
+ long building = failed.getVectorBuildingGeneration();
+ assertEquals(Long.valueOf(active), failed.getVectorCentroidGeneration());
+ assertEquals(count(conn, table), count(conn, index));
+
+ conn.createStatement().execute("ALTER INDEX " + index + " ON " + table + " REBUILD");
+ PTable resumed = index(conn, index);
+ assertEquals("The resumed rebuild completes the same generation", Long.valueOf(building),
+ resumed.getVectorCentroidGeneration());
+ assertFalse(resumed.isVectorRebuildInProgress());
+ assertEquals(0, countCentroids(conn, index, active));
+ assertIndexVerifies(table, index, 40);
+ }
+ }
+
+ /**
+ * Verifies that concurrent rebuild attempts on the same index are rejected when a lock is held.
+ */
+ @Test
+ public void testConcurrentRebuildIsExcluded() throws Exception {
+ String table = "T_CR_" + generateUniqueName();
+ String index = "IDX_CR_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ for (int i = 0; i < 8; i++) {
+ upsert(conn, table, "r" + i, near(i % 4, 0.01f), "l");
+ }
+ conn.commit();
+ assertTrue(CentroidManager.claimRebuild(conn, index, "other", 60000));
+ assertEquals(Outcome.IN_PROGRESS,
+ VectorIndexRebuilder.rebuild(conn.unwrap(PhoenixConnection.class), index, true, "x"));
+ try {
+ conn.createStatement().execute("ALTER INDEX " + index + " ON " + table + " REBUILD");
+ fail();
+ } catch (SQLException e) {
+ assertTrue(e.getMessage(), e.getMessage().contains("IN_PROGRESS"));
+ }
+ CentroidManager.releaseRebuild(conn, index, "other");
+ }
+ }
+
+ /**
+ * Verifies that a drift flagged by reassignment rate alone starts the automatic rebuild it
+ * queues. Reconciliation resets the reassignment counters when it assesses, so the queued rebuild
+ * must act on that assessment rather than reconcile and assess again.
+ */
+ @Test
+ public void testReassignmentDriftRebuilds() throws Exception {
+ String table = "T_RD_" + generateUniqueName();
+ String index = "IDX_RD_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long gen = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 40; i++) {
+ upsert(conn, table, "b" + i, near(i % 4, 0.001f * i), "l");
+ }
+ conn.commit();
+ flushScorecards();
+ // Balanced posting lists, with 20 reassignments among 40 rows (rate 0.5)
+ CentroidManager.adjustScorecard(conn, index, gen,
+ java.util.Arrays.asList(new ScorecardRow(0, 0, 20)));
+ try (PhoenixConnection enabled = connect(true).unwrap(PhoenixConnection.class)) {
+ assertEquals(VectorIndexRebuilder.ReconcileOutcome.RECONCILED,
+ VectorIndexRebuilder.reconcile(enabled, index));
+ List<String> queued = queuedAutomaticRebuilds(conn, index);
+ assertEquals(1, queued.size());
+ assertTrue(queued.get(0), queued.get(0).startsWith("REASSIGN_RATE_EXCEEDED"));
+ assertEquals(0, reassigned(conn, index, gen, 0));
+ // Run the queued rebuild as the rebuild task does
+ assertEquals(Outcome.REBUILT,
+ VectorIndexRebuilder.rebuild(enabled, index, false, queued.get(0)));
+ }
+ long rebuilt = index(conn, index).getVectorCentroidGeneration();
+ assertTrue(rebuilt > gen);
+ assertTrue(CentroidManager.loadGenerationSummary(conn, index, rebuilt).getTriggerReason()
+ .startsWith("REASSIGN_RATE_EXCEEDED"));
+ }
+ }
+
+ /**
+ * Verifies gating conditions for automatic index rebuilds: reconciliation erases counter error
+ * before assessing, and queues a rebuild only for real drift with automatic rebuild enabled; the
+ * queued rebuild rechecks the configuration flag and the minimum rebuild interval.
+ */
+ @Test
+ public void testAutomaticRebuildGates() throws Exception {
+ String table = "T_AG_" + generateUniqueName();
+ String index = "IDX_AG_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long gen = index(conn, index).getVectorCentroidGeneration();
+ // Populate balanced clusters, then inject an artificially skewed scorecard
+ for (int i = 0; i < 40; i++) {
+ upsert(conn, table, "b" + i, near(i % 4, 0.001f * i), "l");
+ }
+ conn.commit();
+ flushScorecards();
+ // Skew centroids 0 and 1, which hold 10 rows each, to 1000 and 0
+ CentroidManager.adjustScorecard(conn, index, gen,
+ java.util.Arrays.asList(new ScorecardRow(0, 990, 0), new ScorecardRow(1, -10, 0)));
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ try (PhoenixConnection enabled = connect(true).unwrap(PhoenixConnection.class)) {
+ // Reconciliation erases the injected skew before assessing, so nothing is queued
+ assertEquals(VectorIndexRebuilder.ReconcileOutcome.RECONCILED,
+ VectorIndexRebuilder.reconcile(enabled, index));
+ assertTrue(queuedAutomaticRebuilds(conn, index).isEmpty());
+ }
+ // Skew data distribution towards centroid 0
+ for (int i = 0; i < 40; i++) {
+ upsert(conn, table, "b" + i, near(0, 0.01f * (i % 10)), "l");
+ }
+ conn.commit();
+ // Drift is assessed and recorded, but automatic rebuild is disabled, so nothing is queued
+ assertEquals(VectorIndexRebuilder.ReconcileOutcome.RECONCILED,
+ VectorIndexRebuilder.reconcile(pconn, index));
+ assertTrue(queuedAutomaticRebuilds(conn, index).isEmpty());
+ String reason = CentroidManager.loadGenerationSummary(conn, index, gen).getTriggerReason();
+ assertTrue(reason, reason.contains("SKEW_RATIO_EXCEEDED: unbounded"));
+ // A queued rebuild that runs once automatic rebuild is disabled does nothing
+ assertEquals(Outcome.DISABLED, VectorIndexRebuilder.rebuild(pconn, index, false, reason));
+ assertEquals(gen, index(conn, index).getVectorCentroidGeneration().longValue());
+
+ try (PhoenixConnection enabled = connect(true).unwrap(PhoenixConnection.class)) {
+ assertEquals(VectorIndexRebuilder.ReconcileOutcome.RECONCILED,
+ VectorIndexRebuilder.reconcile(enabled, index));
+ List<String> queued = queuedAutomaticRebuilds(conn, index);
+ assertEquals(1, queued.size());
+ assertEquals(Outcome.REBUILT,
+ VectorIndexRebuilder.rebuild(enabled, index, false, queued.get(0)));
+ long rebuilt = index(conn, index).getVectorCentroidGeneration();
+ assertTrue(rebuilt > gen);
+ assertTrue(CentroidManager.loadGenerationSummary(conn, index, rebuilt).getTriggerReason()
+ .contains("SKEW_RATIO_EXCEEDED"));
+ Properties soon = props(true);
+ soon.setProperty(QueryServices.VECTOR_DRIFT_SKEW_RATIO_THRESHOLD_ATTRIB, "0.5");
+ soon.setProperty(QueryServices.VECTOR_REBUILD_MIN_INTERVAL_MS_ATTRIB, "86400000");
+ try (PhoenixConnection eager =
+ DriverManager.getConnection(getUrl(), soon).unwrap(PhoenixConnection.class)) {
+ assertEquals(Outcome.TOO_SOON,
+ VectorIndexRebuilder.rebuild(eager, index, false, "SKEW_RATIO_EXCEEDED"));
+ assertEquals(Outcome.REBUILT, VectorIndexRebuilder.rebuild(eager, index, true, "OP"));
+ }
+ }
+ // Ensure multi-generational row keys do not collide across migration passes
+ assertEquals(40, count(conn, index));
+ assertIndexConsistent(table, index);
+ }
+ }
+
+ /** Verifies index rebuild support for functional vector indexes over BSON expressions. */
+ @Test
+ public void testFunctionalIndexRebuild() throws Exception {
+ String table = "T_BR_" + generateUniqueName();
+ String index = "IDX_BR_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ conn.createStatement()
+ .execute("CREATE TABLE " + table + " (ID VARCHAR NOT NULL PRIMARY KEY, DOC BSON)");
+ try (
+ PreparedStatement ps = conn.prepareStatement("UPSERT INTO " + table + " VALUES (?, ?)")) {
+ Random rng = new Random(3);
+ for (int i = 0; i < 60; i++) {
+ float[] v = new float[4];
+ for (int d = 0; d < 4; d++) {
+ v[d] = rng.nextFloat();
+ }
+ ps.setString(1, "d" + i);
+ ps.setObject(2,
+ new BsonDocument("embedding", new BsonBinary(BinaryVector.floatVector(v))));
+ ps.executeUpdate();
+ }
+ }
+ conn.commit();
+ conn.createStatement()
+ .execute("CREATE VECTOR INDEX " + index + " ON " + table
+ + " (BSON_VECTOR_VALUE(DOC, 'embedding', 4)) WITH (algorithm='IVF', metric='L2', lists=4,"
+ + " sample_size=60)");
+ long before = index(conn, index).getVectorCentroidGeneration();
+ conn.createStatement().execute("ALTER INDEX " + index + " ON " + table + " REBUILD");
+ assertNotEquals(before, index(conn, index).getVectorCentroidGeneration().longValue());
+ assertEquals(60, count(conn, index));
+ assertIndexVerifies(table, index, 60);
+ }
+ }
+
+ /**
+ * Verifies asynchronous ALTER INDEX REBUILD ASYNC task creation and background execution by the
+ * task handler.
+ */
+ @Test
+ public void testAsyncRebuildRunsAsTask() throws Exception {
+ String table = "T_AR_" + generateUniqueName();
+ String index = "IDX_AR_" + generateUniqueName();
+ try (Connection conn = connect(false)) {
+ setupTableAndKnownCentroids(conn, table, index);
+ long before = index(conn, index).getVectorCentroidGeneration();
+ for (int i = 0; i < 20; i++) {
+ upsert(conn, table, "r" + i, near(i % 4, 0.01f * i), "l");
+ }
+ conn.commit();
+ conn.createStatement().execute("ALTER INDEX " + index + " ON " + table + " REBUILD ASYNC");
+ List<Task.TaskRecord> tasks = Task.queryTaskTable(conn, null, null, index,
+ PTable.TaskType.VECTOR_INDEX_REBUILD, null, null);
+ assertEquals(1, tasks.size());
+ assertTrue(tasks.get(0).getData(), tasks.get(0).getData().contains("\"manual\":true"));
+ runTaskSweep();
+ assertTrue(index(conn, index).getVectorCentroidGeneration() > before);
+ assertEquals(PTable.TaskStatus.COMPLETED.toString(),
+ Task
+ .queryTaskTable(conn, null, null, index, PTable.TaskType.VECTOR_INDEX_REBUILD, null, null)
+ .get(0).getStatus());
+ assertIndexVerifies(table, index, 20);
+ }
+ }
+
+ private static float[] unbox(Float[] v) {
+ float[] f = new float[v.length];
+ for (int i = 0; i < v.length; i++) {
+ f[i] = v[i];
+ }
+ return f;
+ }
+
+ private static List<String> topK(Connection conn, String table, String hint, float[] q, int k)
+ throws SQLException {
+ List<String> ids = new ArrayList<>();
+ try (PreparedStatement ps = conn.prepareStatement(
+ "SELECT " + hint + " ID FROM " + table + " ORDER BY L2_DISTANCE(V, ?) LIMIT " + k)) {
+ ps.setArray(1, conn.createArrayOf("FLOAT", box(q)));
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ ids.add(rs.getString(1));
+ }
+ }
+ }
+ return ids;
+ }
+
+ private static String explain(Connection conn, String table, float[] q) throws SQLException {
+ StringBuilder sb = new StringBuilder();
+ try (PreparedStatement ps = conn.prepareStatement(
+ "EXPLAIN SELECT ID FROM " + table + " ORDER BY L2_DISTANCE(V, ?) LIMIT 3")) {
+ ps.setArray(1, conn.createArrayOf("FLOAT", box(q)));
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ sb.append(rs.getString(1)).append('\n');
+ }
+ }
+ }
+ return sb.toString();
+ }
+
+ private static Float[] box(float[] q) {
+ Float[] b = new Float[q.length];
+ for (int i = 0; i < q.length; i++) {
+ b[i] = q[i];
+ }
+ return b;
+ }
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexTestUtil.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexTestUtil.java
index 89c26fffb2..0ff54b3096 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexTestUtil.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexTestUtil.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.end2end;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CENTROID_ID;
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.SYSTEM_VECTOR_CENTROID_NAME;
@@ -123,7 +124,24 @@ class VectorIndexTestUtil {
.findCounter(PhoenixIndexToolJobCounters.BEFORE_REBUILD_MISSING_INDEX_ROW_COUNT).getValue());
}
- /** Populates uniform random vector data across optional tenant partitions. */
+ /**
+ * Runs {@link IndexTool} verification and asserts that no index rows are invalid or missing.
+ * Unlike {@link #assertIndexVerifies}, valid row count is not compared against data table size.
+ */
+ static void assertIndexConsistent(String tableName, String indexName) throws Exception {
+ IndexTool tool = IndexToolIT.runIndexTool(false, null, tableName, indexName, null, 0,
+ IndexTool.IndexVerifyType.ONLY);
+ Counters counters = tool.getJob().getCounters();
+ assertEquals("invalid index rows", 0, counters
+ .findCounter(PhoenixIndexToolJobCounters.BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT).getValue());
+ assertEquals("missing index rows", 0, counters
+ .findCounter(PhoenixIndexToolJobCounters.BEFORE_REBUILD_MISSING_INDEX_ROW_COUNT).getValue());
+ }
+
+ /**
+ * Upserts {@code rows} random vectors in [0, 10)^4 with ID {@code row_NNN} and a LABEL, spreading
+ * them over {@code tenants} when {@code tenantCol} is not null.
+ */
static float[][] loadRandomVectors(Connection conn, String tableName, String tenantCol,
String[] tenants, int rows, long seed) throws SQLException {
Random rng = new Random(seed);
@@ -442,8 +460,9 @@ class VectorIndexTestUtil {
static int countCentroids(Connection conn, String indexName, Long generation)
throws SQLException {
- String sql = "SELECT COUNT(*) FROM " + SYSTEM_VECTOR_CENTROID_NAME + " WHERE " + INDEX_NAME
- + " = ?" + (generation == null ? "" : " AND " + GENERATION_ID + " = ?");
+ String sql =
+ "SELECT COUNT(*) FROM " + SYSTEM_VECTOR_CENTROID_NAME + " WHERE " + INDEX_NAME + " = ? AND "
+ + CENTROID_ID + " >= 0" + (generation == null ? "" : " AND " + GENERATION_ID + " = ?");
try (PreparedStatement ps = conn.prepareStatement(sql)) {
ps.setString(1, indexName);
if (generation != null) {
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/index/vector/VectorIndexRebuilderTestHooks.java b/phoenix-core/src/it/java/org/apache/phoenix/index/vector/VectorIndexRebuilderTestHooks.java
new file mode 100644
index 0000000000..0f27ff0b48
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/index/vector/VectorIndexRebuilderTestHooks.java
@@ -0,0 +1,45 @@
+/*
+ * 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 org.apache.phoenix.schema.PTable;
+
+/** Test hooks for mid-migration states during index rebuild and for scorecard reconciliation. */
+public final class VectorIndexRebuilderTestHooks {
+
+ /** Invoked between the first and second scan passes of a rebuild migration. */
+ public interface Hook {
+ void migrating(PTable index, long buildingGeneration) throws Exception;
+ }
+
+ private VectorIndexRebuilderTestHooks() {
+ }
+
+ public static void setMigrationHook(Hook hook) {
+ VectorIndexRebuilder.setMigrationHookForTesting(hook == null ? null : hook::migrating);
+ }
+
+ /** Invoked after reconciliation reads the scorecard and before it writes corrections. */
+ public interface ReconcileHook {
+ void loaded(String indexName, long generation) throws Exception;
+ }
+
+ public static void setReconcileHook(ReconcileHook hook) {
+ VectorIndexScorecard.setReconcileHookForTesting(hook == null ? null : hook::loaded);
+ }
+}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/cache/VectorCentroidCacheTest.java b/phoenix-core/src/test/java/org/apache/phoenix/cache/VectorCentroidCacheTest.java
index 9de9e0a689..2719120dd1 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/cache/VectorCentroidCacheTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/cache/VectorCentroidCacheTest.java
@@ -237,4 +237,16 @@ public class VectorCentroidCacheTest {
cache.getIfPresent("A", 3L);
assertEquals(2, cache.size());
}
+
+ /** Verifies centroid assignment and indexing when IDs start from a non-zero base ID. */
+ @Test
+ public void testCentroidIdsStartAtFirstId() {
+ CachedCentroids c = new CachedCentroids(
+ centroids(new float[] { 0, 0 }, new float[] { 10, 0 }, new float[] { 0, 10 }),
+ DistanceMetric.L2, 7);
+ assertEquals(7, c.getFirstId());
+ assertEquals(8, c.assign(new float[] { 9.5f, 0.5f }));
+ assertArrayEquals(new int[] { 8, 7 }, c.nearest(new float[] { 9.5f, 0.5f }, 2));
+ assertArrayEquals(new float[] { 0, 10 }, c.getCentroid(9), 0f);
+ }
}
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 d3b81df4f0..d75a4892e3 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
@@ -207,4 +207,15 @@ public class VectorIndexScanPlanTest extends BaseConnectionlessQueryTest {
HintNode invalid = new HintNode("/*+ VECTOR_PROBE_COUNT(abc) */");
assertEquals(7, VectorIndexScanPlan.getHintInt(invalid, Hint.VECTOR_PROBE_COUNT, 7));
}
+
+ @Test
+ public void testInterleaveAlternatesGenerationRankings() {
+ // Any prefix of 2n holds the n nearest of each generation, and a longer ranking keeps its tail
+ assertArrayEquals(new int[] { 0, 4, 1, 5, 2, 6 },
+ VectorIndexScanPlan.interleave(new int[] { 0, 1, 2 }, new int[] { 4, 5, 6 }));
+ assertArrayEquals(new int[] { 3, 9, 1, 8, 7 },
+ VectorIndexScanPlan.interleave(new int[] { 3, 1 }, new int[] { 9, 8, 7 }));
+ assertArrayEquals(new int[] { 2, 0 },
+ VectorIndexScanPlan.interleave(new int[] { 2, 0 }, new int[0]));
+ }
}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/index/vector/VectorIndexRebuilderTest.java b/phoenix-core/src/test/java/org/apache/phoenix/index/vector/VectorIndexRebuilderTest.java
new file mode 100644
index 0000000000..fe35b57bf8
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/index/vector/VectorIndexRebuilderTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.index.vector.VectorIndexRebuilder.isDue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.phoenix.util.JacksonUtil;
+import org.junit.Test;
+
+public class VectorIndexRebuilderTest {
+
+ private static final long INTERVAL = 86400000L;
+ private static final long NOW = 1_000_000_000_000L;
+
+ @Test
+ public void testNeverReconciledIsDue() {
+ assertTrue(isDue(null, INTERVAL, NOW));
+ }
+
+ @Test
+ public void testNotDueWithinInterval() {
+ assertFalse(isDue(NOW, INTERVAL, NOW));
+ assertFalse(isDue(NOW - INTERVAL + 1, INTERVAL, NOW));
+ }
+
+ @Test
+ public void testDueAtInterval() {
+ assertTrue(isDue(NOW - INTERVAL, INTERVAL, NOW));
+ assertTrue(isDue(NOW - 10 * INTERVAL, INTERVAL, NOW));
+ }
+
+ @Test
+ public void testZeroIntervalIsAlwaysDue() {
+ assertTrue(isDue(NOW, 0, NOW));
+ }
+
+ /** Verifies that future timestamps (e.g. from clock skew) do not trigger reconciliation. */
+ @Test
+ public void testFutureUpdateIsNotDue() {
+ assertFalse(isDue(NOW + INTERVAL, INTERVAL, NOW));
+ assertFalse(isDue(NOW + 1, 0, NOW));
+ }
+
+ @Test
+ public void testRebuildTaskData() throws Exception {
+ JsonNode data = JacksonUtil.getObjectReader()
+ .readTree(VectorIndexRebuilder.rebuildTaskData(false, "SKEW_RATIO_EXCEEDED: \"x\""));
+ assertFalse(data.path("manual").asBoolean(true));
+ assertEquals("SKEW_RATIO_EXCEEDED: 'x'", data.path("reason").asText());
+ }
+}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/index/vector/VectorIndexScorecardTest.java b/phoenix-core/src/test/java/org/apache/phoenix/index/vector/VectorIndexScorecardTest.java
new file mode 100644
index 0000000000..90ec2d9bcf
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/index/vector/VectorIndexScorecardTest.java
@@ -0,0 +1,138 @@
+/*
+ * 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.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.phoenix.index.vector.VectorIndexScorecard.Assessment;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.Test;
+
+public class VectorIndexScorecardTest {
+
+ private static List<ScorecardRow> rows(long... sizes) {
+ List<ScorecardRow> rows = new ArrayList<>();
+ for (int i = 0; i < sizes.length; i++) {
+ rows.add(new ScorecardRow(i, sizes[i], 0));
+ }
+ return rows;
+ }
+
+ private static ReadOnlyProps conf() {
+ return conf(new HashMap<>());
+ }
+
+ private static ReadOnlyProps conf(Map<String, String> props) {
+ props.put(QueryServices.VECTOR_DRIFT_MIN_POPULATION_ATTRIB, "1000");
+ return new ReadOnlyProps(props);
+ }
+
+ @Test
+ public void testBalancedIndexIsNotDrifted() {
+ Assessment assessment = VectorIndexScorecard.assess(rows(500, 500, 500, 500), conf());
+ assertFalse(assessment.isDrifted());
+ assertNull(assessment.getReason());
+ assertEquals(1.0, assessment.getSkewRatio(), 1e-9);
+ assertEquals(0.0, assessment.getSizeCv(), 1e-9);
+ assertEquals(0.0, assessment.getEmptyCentroidFraction(), 1e-9);
+ assertEquals(2000, assessment.getPopulation());
+ }
+
+ @Test
+ public void testSkewRatioIsTakenAgainstTheMedian() {
+ // 9 clusters of size 100 and 1 cluster of size 9100 (median: 100, mean: 1000)
+ long[] sizes = new long[10];
+ java.util.Arrays.fill(sizes, 100);
+ sizes[9] = 9100;
+ Assessment assessment = VectorIndexScorecard.assess(rows(sizes), conf());
+ assertEquals(91.0, assessment.getSkewRatio(), 1e-9);
+ assertTrue(assessment.isDrifted());
+ assertTrue(assessment.getReason(), assessment.getReason().contains("SKEW_RATIO_EXCEEDED"));
+ }
+
+ /**
+ * Verifies that when the median cluster size is zero, skew ratio evaluates to unbounded.
+ */
+ @Test
+ public void testZeroMedianIsUnboundedSkew() {
+ Assessment assessment = VectorIndexScorecard.assess(rows(0, 0, 0, 1000), conf());
+ assertTrue(Double.isInfinite(assessment.getSkewRatio()));
+ assertTrue(assessment.isDrifted());
+ assertTrue(assessment.getReason(),
+ assessment.getReason().contains("SKEW_RATIO_EXCEEDED: unbounded"));
+ }
+
+ @Test
+ public void testEmptyIndexHasNoSkew() {
+ Assessment assessment = VectorIndexScorecard.assess(rows(0, 0, 0, 0), conf());
+ assertEquals(0.0, assessment.getSkewRatio(), 1e-9);
+ assertFalse(assessment.isDrifted());
+ }
+
+ @Test
+ public void testSmallIndexIsNotAssessed() {
+ Assessment assessment = VectorIndexScorecard.assess(rows(0, 0, 0, 999), conf());
+ assertTrue(Double.isInfinite(assessment.getSkewRatio()));
+ assertFalse("below the minimum population", assessment.isDrifted());
+ }
+
+ @Test
+ public void testEmptyCentroidFraction() {
+ // Half the clusters empty and half populated (median: 500)
+ Assessment assessment = VectorIndexScorecard.assess(rows(0, 0, 1000, 1000), conf());
+ assertEquals(0.5, assessment.getEmptyCentroidFraction(), 1e-9);
+ assertTrue(assessment.getReason(),
+ assessment.getReason().contains("EMPTY_CENTROID_FRACTION_EXCEEDED"));
+ }
+
+ @Test
+ public void testReassignmentRate() {
+ List<ScorecardRow> rows = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ rows.add(new ScorecardRow(i, 500, 150));
+ }
+ Assessment assessment = VectorIndexScorecard.assess(rows, conf());
+ assertEquals(0.3, assessment.getReassignmentRate(), 1e-9);
+ assertEquals("REASSIGN_RATE_EXCEEDED: 0.30 > 0.20", assessment.getReason());
+ }
+
+ @Test
+ public void testSkewRatioThresholdIsConfigurable() {
+ Map<String, String> props = new HashMap<>();
+ props.put(QueryServices.VECTOR_DRIFT_SKEW_RATIO_THRESHOLD_ATTRIB, "100");
+ ReadOnlyProps conf = conf(props);
+ long[] sizes = new long[10];
+ java.util.Arrays.fill(sizes, 100);
+ sizes[9] = 9100;
+ Assessment assessment = VectorIndexScorecard.assess(rows(sizes), conf);
+ assertFalse(assessment.getReason(), assessment.getReason().contains("SKEW_RATIO_EXCEEDED"));
+ }
+
+ @Test
+ public void testNoCentroids() {
+ assertFalse(VectorIndexScorecard.assess(new ArrayList<>(), conf()).isDrifted());
+ }
+}