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/09 05:10:06 UTC
(phoenix) 07/16: PHOENIX-7998 Phase 7: Server-side write pipeline for IVF
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 87970aa53e631273fe081e35a19fbd1496a22fdb
Author: Andrew Purtell <ap...@apache.org>
AuthorDate: Sat Sep 19 22:27:33 2026 -0700
PHOENIX-7998 Phase 7: Server-side write pipeline for IVF
Co-Authored-By: Claude Opus 5.5 (1M context) <no...@anthropic.com>
---
.../hbase/ipc/controller/ServerRpcController.java | 4 +
.../apache/phoenix/cache/VectorCentroidCache.java | 47 +-
.../phoenix/compile/CreateIndexCompiler.java | 11 +-
.../org/apache/phoenix/compile/DeleteCompiler.java | 22 +-
.../apache/phoenix/exception/SQLExceptionCode.java | 2 +
.../org/apache/phoenix/index/IndexMaintainer.java | 174 +++-
.../phoenix/index/vector/CentroidManager.java | 12 +
.../org/apache/phoenix/schema/MetaDataClient.java | 34 +-
.../java/org/apache/phoenix/util/IndexUtil.java | 15 +-
.../src/main/protobuf/ServerCachingService.proto | 3 +
.../phoenix/hbase/index/IndexRegionObserver.java | 24 +-
.../apache/phoenix/mapreduce/index/IndexTool.java | 18 +
.../org/apache/phoenix/end2end/VectorIndexIT.java | 886 ++++++++++++++++++++-
.../controller/ServerRpcControllerFactoryTest.java | 58 ++
.../apache/phoenix/index/IndexMaintainerTest.java | 245 ++++++
.../phoenix/index/vector/CentroidManagerTest.java | 11 +
16 files changed, 1521 insertions(+), 45 deletions(-)
diff --git a/phoenix-core-client/src/main/java/org/apache/hadoop/hbase/ipc/controller/ServerRpcController.java b/phoenix-core-client/src/main/java/org/apache/hadoop/hbase/ipc/controller/ServerRpcController.java
index 737aa8ee06..c813de8364 100644
--- a/phoenix-core-client/src/main/java/org/apache/hadoop/hbase/ipc/controller/ServerRpcController.java
+++ b/phoenix-core-client/src/main/java/org/apache/hadoop/hbase/ipc/controller/ServerRpcController.java
@@ -42,6 +42,7 @@ class ServerRpcController extends DelegatingHBaseRpcController {
.add(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME)
.add(PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME)
.add(PhoenixDatabaseMetaData.SYSTEM_CHILD_LINK_NAME)
+ .add(PhoenixDatabaseMetaData.SYSTEM_VECTOR_CENTROID_NAME)
.add(SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, true)
.getNameAsString())
.add(SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES, true)
@@ -52,6 +53,9 @@ class ServerRpcController extends DelegatingHBaseRpcController {
.getNameAsString())
.add(SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CHILD_LINK_NAME_BYTES, true)
.getNameAsString())
+ .add(SchemaUtil
+ .getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_VECTOR_CENTROID_NAME_BYTES, true)
+ .getNameAsString())
.build();
ServerRpcController(HBaseRpcController delegate, Configuration conf) {
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 0e643c5955..8fd318adb4 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
@@ -28,12 +28,14 @@ import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.index.vector.CentroidManager;
import org.apache.phoenix.index.vector.KMeansTrainer;
+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.types.PDataType;
import org.apache.phoenix.schema.types.PVectorDouble;
import org.apache.phoenix.schema.types.PVectorFloat;
+import org.apache.phoenix.util.QueryUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +54,8 @@ public final class VectorCentroidCache {
private static final Logger LOGGER = LoggerFactory.getLogger(VectorCentroidCache.class);
private static volatile VectorCentroidCache instance;
+ /** The server configuration that index maintenance uses to load centroids on a cache miss. */
+ private static volatile Configuration serverConf;
private final Cache<Key, CachedCentroids> cache;
private final long maxBytes;
@@ -102,6 +106,43 @@ public final class VectorCentroidCache {
}
}
+ /**
+ * Sets the server configuration that {@link #getForWrite} uses to load centroids on a cache miss
+ * without a caller connection. Only the first call sets the value.
+ */
+ public static void setServerConfiguration(Configuration conf) {
+ if (serverConf == null) {
+ serverConf = conf;
+ }
+ }
+
+ /**
+ * Returns the centroids that vector index maintenance uses for writes. A cache miss loads them
+ * through {@code conn}, or through an internal server connection if {@code conn} is null.
+ * @throws SQLException if no centroids are recorded for the generation, if the load fails, or if
+ * {@code conn} is null and no server configuration is set
+ */
+ public static CachedCentroids getForWrite(Connection conn, String indexName, long generation,
+ DistanceMetric metric) throws SQLException {
+ VectorCentroidCache cache = instance;
+ CachedCentroids centroids = cache == null ? null : cache.getIfPresent(indexName, generation);
+ if (centroids != null) {
+ return centroids;
+ }
+ if (conn != null) {
+ return getInstance(conn.unwrap(PhoenixConnection.class).getQueryServices().getConfiguration())
+ .get(conn, indexName, generation, metric);
+ }
+ Configuration conf = serverConf;
+ if (conf == null) {
+ throw new SQLException("Centroids of vector index " + indexName + " generation " + generation
+ + " are not loaded and no server connection is available");
+ }
+ try (Connection serverConn = QueryUtil.getConnectionOnServer(conf)) {
+ return getInstance(conf).get(serverConn, indexName, generation, metric);
+ }
+ }
+
/** Returns the cached centroids of the index generation, or null on a miss. It does not load. */
public CachedCentroids getIfPresent(String indexName, long generation) {
return cache.getIfPresent(new Key(indexName, generation));
@@ -145,8 +186,10 @@ public final class VectorCentroidCache {
throws SQLException {
List<float[]> centroids = CentroidManager.loadCentroids(conn, key.indexName, key.generation);
if (centroids.isEmpty()) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.INVALID_VECTOR_INDEX_PARAMS).setMessage(
- "No centroids recorded for vector index " + key.indexName + " generation " + key.generation)
+ // A generation without recorded centroids shows that the caller has 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);
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/CreateIndexCompiler.java b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/CreateIndexCompiler.java
index 18e0262490..eb2157bb75 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/CreateIndexCompiler.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/CreateIndexCompiler.java
@@ -299,8 +299,10 @@ public class CreateIndexCompiler {
}
/**
- * Validates vector index constraints including base table type, indexed expression type and
- * dimensionality, distance metric, and algorithm parameters.
+ * Validates a vector index definition. The data table must not be a view or a transactional
+ * table. The index must have one ascending vector expression with a dimension known at compile
+ * time. The metric must be supported. The algorithm must be IVF, with {@code lists > 0} and
+ * {@code sample_size >= lists}. The index consistency must not be eventual.
*/
private void verifyVectorIndex(CreateIndexStatement create, StatementContext context)
throws SQLException {
@@ -309,6 +311,11 @@ public class CreateIndexCompiler {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.INVALID_VECTOR_INDEX_PARAMS)
.setMessage("Vector indexes are not supported on views").build().buildException();
}
+ if (dataTable.isTransactional()) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.VECTOR_INDEX_ON_TRANSACTIONAL_TABLE)
+ .setSchemaName(dataTable.getSchemaName().getString())
+ .setTableName(dataTable.getTableName().getString()).build().buildException();
+ }
List<Pair<ParseNode, SortOrder>> pairs =
create.getIndexConstraint().getParseNodeAndSortOrderList();
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 4e9907e76c..218ed80981 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
@@ -248,6 +248,11 @@ public class DeleteCompiler {
}
}
+ if (maintainers != null) {
+ for (IndexMaintainer maintainer : maintainers) {
+ maintainer.loadCentroids(connection);
+ }
+ }
byte[][] viewConstants = IndexUtil.getViewConstants(dataTable);
int rowCount = 0;
while (rs.next()) {
@@ -272,12 +277,21 @@ public class DeleteCompiler {
if (table.getType() == PTableType.INDEX) {
otherRowKeyPtr.set(scannedIndexMaintainer.buildDataRowKey(rowKeyPtr, viewConstants));
if (otherTable.getType() == PTableType.INDEX) {
- otherRowKeyPtr.set(maintainers[i].buildRowKey(getter, otherRowKeyPtr, null, null,
- rs.getCurrentRow().getValue(0).getTimestamp()));
+ byte[] otherRowKey = maintainers[i].buildRowKey(getter, otherRowKeyPtr, null, null,
+ rs.getCurrentRow().getValue(0).getTimestamp());
+ if (otherRowKey == null) {
+ continue;
+ }
+ otherRowKeyPtr.set(otherRowKey);
}
} else {
- otherRowKeyPtr.set(maintainers[i].buildRowKey(getter, rowKeyPtr, null, null,
- rs.getCurrentRow().getValue(0).getTimestamp()));
+ // The row key is null if the row has no index row, for example if its vector is null
+ byte[] otherRowKey = maintainers[i].buildRowKey(getter, rowKeyPtr, null, null,
+ rs.getCurrentRow().getValue(0).getTimestamp());
+ if (otherRowKey == null) {
+ continue;
+ }
+ otherRowKeyPtr.set(otherRowKey);
}
otherMutations.get(i).put(otherRowKeyPtr,
new RowMutationState(PRow.DELETE_MARKER, 0,
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core-client/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index 0abafeefbd..1b486ce453 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -126,6 +126,8 @@ public enum SQLExceptionCode {
UNSUPPORTED_VECTOR_INDEX_ALGORITHM(309, "23107", "Unsupported vector index algorithm: "),
UNSUPPORTED_VECTOR_DISTANCE_METRIC(310, "23108", "Unsupported vector distance metric: "),
INVALID_VECTOR_INDEX_PARAMS(311, "23109", "Invalid vector index parameters: "),
+ VECTOR_INDEX_ON_TRANSACTIONAL_TABLE(312, "23110",
+ "Vector indexes are not supported on transactional tables."),
/**
* Invalid Cursor State (errorcode 04, sqlstate 24)
*/
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 533ddadb10..ee3cf88ffc 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
@@ -27,6 +27,7 @@ import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -54,6 +55,8 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
+import org.apache.phoenix.cache.VectorCentroidCache;
+import org.apache.phoenix.cache.VectorCentroidCache.CachedCentroids;
import org.apache.phoenix.compat.hbase.ByteStringer;
import org.apache.phoenix.compile.ColumnResolver;
import org.apache.phoenix.compile.FromCompiler;
@@ -76,8 +79,11 @@ import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.hbase.index.util.GenericKeyValueBuilder;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
+import org.apache.phoenix.index.vector.CentroidManager;
+import org.apache.phoenix.index.vector.VectorIndexTrainer;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixStatement;
+import org.apache.phoenix.optimize.DistanceMetric;
import org.apache.phoenix.parse.FunctionParseNode;
import org.apache.phoenix.parse.ParseNode;
import org.apache.phoenix.parse.SQLParser;
@@ -109,6 +115,7 @@ import org.apache.phoenix.schema.tuple.ValueGetterTuple;
import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PBoolean;
import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PInteger;
import org.apache.phoenix.schema.types.PVarbinaryEncoded;
import org.apache.phoenix.transaction.PhoenixTransactionProvider.Feature;
import org.apache.phoenix.util.BitSet;
@@ -167,6 +174,10 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
*/
public static boolean sendIndexMaintainer(PTable index) {
PIndexState indexState = index.getIndexState();
+ if (index.isVectorIndex() && index.getVectorCentroidGeneration() == null) {
+ // An untrained vector index has no centroids and keeps no index rows, so send no maintainer
+ return false;
+ }
return !(indexState.isDisabled() || PIndexState.PENDING_ACTIVE == indexState);
}
@@ -476,6 +487,14 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
private boolean isCDCIndex;
private IndexConsistency indexConsistency;
+ // Vector index attributes. The first indexed slot of the row key holds the vector centroid ID.
+ private String vectorAlgorithm;
+ private DistanceMetric distanceMetric;
+ // The maintainer keeps the centroid models for writes during its full lifetime. Thus a cache
+ // eviction cannot cause a reload for each row, or a failure on a client with no server connection
+ private CachedCentroids activeCentroids;
+ private Long centroidGeneration;
+
protected IndexMaintainer(RowKeySchema dataRowKeySchema, boolean isDataTableSalted) {
this.dataRowKeySchema = dataRowKeySchema;
this.isDataTableSalted = isDataTableSalted;
@@ -501,6 +520,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
this.encodingScheme = index.getEncodingScheme();
this.isCDCIndex = CDCUtil.isCDCIndex(index);
this.indexConsistency = index.getIndexConsistency();
+ if (index.isVectorIndex()) {
+ this.vectorAlgorithm = index.getVectorIndexAlgorithm();
+ this.distanceMetric = DistanceMetric.fromString(index.getVectorDistanceMetric());
+ this.centroidGeneration = index.getVectorCentroidGeneration();
+ }
// null check for b/w compatibility
this.encodingScheme = index.getEncodingScheme() == null
@@ -530,6 +554,10 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
int indexedExpressionCount = 0;
for (int i = indexPosOffset; i < index.getPKColumns().size(); i++) {
PColumn indexColumn = index.getPKColumns().get(i);
+ if (isCentroidColumn(index, indexColumn)) {
+ indexedExpressionCount++;
+ continue;
+ }
String indexColumnName = indexColumn.getName().getString();
String dataFamilyName = IndexUtil.getDataColumnFamilyName(indexColumnName);
String dataColumnName = IndexUtil.getDataColumnName(indexColumnName);
@@ -602,7 +630,10 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
UDFParseNodeVisitor visitor = new UDFParseNodeVisitor();
for (int i = indexPosOffset; i < index.getPKColumns().size(); i++) {
PColumn indexColumn = index.getPKColumns().get(i);
- String expressionStr = IndexUtil.getIndexColumnExpressionStr(indexColumn);
+ // The centroid ID comes from the indexed vector, so the slot uses the vector expression
+ String expressionStr = isCentroidColumn(index, indexColumn)
+ ? VectorIndexTrainer.getIndexedVectorColumn(index).getExpressionStr()
+ : IndexUtil.getIndexColumnExpressionStr(indexColumn);
try {
ParseNode parseNode = SQLParser.parseCondition(expressionStr);
parseNode.accept(visitor);
@@ -633,7 +664,9 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
if (expressionIndexCompiler.getColumnRef() != null) {
// get the column of the data column that corresponds to this index column
- PColumn column = IndexUtil.getDataColumn(dataTable, indexColumn.getName().getString());
+ PColumn column = isCentroidColumn(index, indexColumn)
+ ? expressionIndexCompiler.getColumnRef().getColumn()
+ : IndexUtil.getDataColumn(dataTable, indexColumn.getName().getString());
boolean isPKColumn = SchemaUtil.isPKColumn(column);
if (isPKColumn) {
int dataPkPos = dataTable.getPKColumns().indexOf(column)
@@ -641,8 +674,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
this.rowKeyMetaData.setIndexPkPosition(dataPkPos, indexPos);
indexedColumnsInfo.add(new Pair<>((String) null, column.getName().getString()));
} else {
- indexColByteSize += column.getDataType().isFixedWidth()
- ? SchemaUtil.getFixedByteSize(column)
+ indexColByteSize += isCentroidColumn(index, indexColumn) ? Bytes.SIZEOF_INT
+ : column.getDataType().isFixedWidth() ? SchemaUtil.getFixedByteSize(column)
: ValueSchema.ESTIMATED_VARIABLE_LENGTH_SIZE;
try {
// Surround constant with cast so that we can still know the original type. Otherwise,
@@ -660,8 +693,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
}
} else {
- indexColByteSize += expression.getDataType().isFixedWidth()
- ? SchemaUtil.getFixedByteSize(expression)
+ indexColByteSize += isCentroidColumn(index, indexColumn) ? Bytes.SIZEOF_INT
+ : expression.getDataType().isFixedWidth() ? SchemaUtil.getFixedByteSize(expression)
: ValueSchema.ESTIMATED_VARIABLE_LENGTH_SIZE;
this.indexedExpressions.add(expression);
KeyValueExpressionVisitor kvVisitor = new KeyValueExpressionVisitor() {
@@ -845,7 +878,25 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
dataColumnType = expression.getDataType();
dataSortOrder = expression.getSortOrder();
isNullable = expression.isNullable();
- if (expression instanceof PartitionIdFunction) {
+ if (i == 0 && isVectorIndex()) {
+ if (centroidGeneration == null) {
+ // An untrained index has no centroids for assignment, so it keeps no index rows
+ return null;
+ }
+ // Rows lacking vector data produce no index entries
+ if (
+ valueGetter == null
+ || !expression.evaluate(new ValueGetterTuple(valueGetter, ts), ptr)
+ || ptr.getLength() == 0
+ ) {
+ return null;
+ }
+ ptr.set(PInteger.INSTANCE.toBytes(
+ getCentroids().assign(ptr.get(), ptr.getOffset(), ptr.getLength(), dataColumnType)));
+ dataColumnType = PInteger.INSTANCE;
+ dataSortOrder = SortOrder.ASC;
+ isNullable = false;
+ } else if (expression instanceof PartitionIdFunction) {
if (i != 0) {
throw new DoNotRetryIOException(
"PARTITION_ID() has to be the prefix " + "of the index row key!");
@@ -1123,6 +1174,10 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
*/
public boolean shouldPrepareIndexMutations(Put dataRowState) {
+ if (isVectorIndex() && getIndexedVector(dataRowState) == null) {
+ // A row with a null vector has no index row and gets no index mutations
+ return false;
+ }
if (getIndexWhere() == null) {
// It is a full index and the index row should be prepared.
return true;
@@ -1237,7 +1292,15 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
boolean isNullableToBe;
Integer maxLengthToBe;
Integer scaleToBe;
- if (indexField == null) {
+ if (indexField == null && isVectorIndex() && i == 0) {
+ // The first indexed column of the index row key is the centroid ID, a non-null INTEGER
+ expressionItr.next();
+ isNullableToBe = false;
+ dataTypeToBe = PInteger.INSTANCE;
+ sortOrderToBe = SortOrder.ASC;
+ maxLengthToBe = null;
+ scaleToBe = null;
+ } else if (indexField == null) {
Expression e = expressionItr.next();
isNullableToBe = e.isNullable();
dataTypeToBe = IndexUtil.getIndexColumnDataType(isNullableToBe, e.getDataType());
@@ -1324,6 +1387,9 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
boolean verified, byte[] encodedRegionName) throws IOException {
byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey,
ts, encodedRegionName);
+ if (indexRowKey == null) {
+ return null;
+ }
return buildUpdateMutation(kvBuilder, valueGetter, dataRowKeyPtr, ts, regionStartKey,
regionEndKey, indexRowKey, this.getEmptyKeyValueFamily(), coveredColumnsMap,
indexEmptyKeyValueRef, indexWALDisabled, dataImmutableStorageScheme, immutableStorageScheme,
@@ -1724,6 +1790,9 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
byte[] regionStartKey, byte[] regionEndKey, byte[] encodedRegionName) throws IOException {
byte[] indexRowKey = this.buildRowKey(oldState, dataRowKeyPtr, regionStartKey, regionEndKey, ts,
encodedRegionName);
+ if (indexRowKey == null) {
+ return null;
+ }
// Delete the entire row if any of the indexed columns changed
DeleteType deleteType = null;
if (
@@ -2076,6 +2145,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
maintainer.nDataTableSaltBuckets =
proto.hasDataTableSaltBuckets() ? proto.getDataTableSaltBuckets() : -1;
+ if (proto.hasVectorAlgorithm()) {
+ maintainer.vectorAlgorithm = proto.getVectorAlgorithm();
+ maintainer.distanceMetric = DistanceMetric.fromString(proto.getDistanceMetric());
+ maintainer.centroidGeneration = proto.getCentroidGeneration();
+ }
maintainer.initCachedState();
return maintainer;
}
@@ -2235,6 +2309,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
if (maintainer.isDataTableSalted) {
builder.setDataTableSaltBuckets(maintainer.nDataTableSaltBuckets);
}
+ if (maintainer.isVectorIndex()) {
+ builder.setVectorAlgorithm(maintainer.vectorAlgorithm);
+ builder.setDistanceMetric(maintainer.distanceMetric.name());
+ builder.setCentroidGeneration(maintainer.centroidGeneration);
+ }
return builder.build();
}
@@ -2622,6 +2701,85 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
return indexConsistency;
}
+ public boolean isVectorIndex() {
+ return vectorAlgorithm != null;
+ }
+
+ public DistanceMetric getDistanceMetric() {
+ return distanceMetric;
+ }
+
+ public Long getCentroidGeneration() {
+ return centroidGeneration;
+ }
+
+ /**
+ * Resolves the centroids of this vector index generation prior to row lock acquisition, from the
+ * cache or, if not cached, via {@code conn} or internal server connection, and holds them for
+ * this maintainer's lifetime. An untrained index has none.
+ */
+ public void loadCentroids(Connection conn) throws SQLException {
+ if (!isVectorIndex() || centroidGeneration == null) {
+ // A non-vector index or an untrained vector index has no centroids to load
+ return;
+ }
+ if (activeCentroids == null) {
+ activeCentroids = loadCentroids(conn, centroidGeneration);
+ }
+ }
+
+ private CachedCentroids loadCentroids(Connection conn, long generation) throws SQLException {
+ // An index that a view inherits uses the centroids of the parent index
+ return VectorCentroidCache.getForWrite(conn,
+ CentroidManager.getCentroidIndexName(logicalIndexName), generation, distanceMetric);
+ }
+
+ private CachedCentroids getCentroids() {
+ try {
+ loadCentroids(null);
+ } catch (SQLException e) {
+ throw new IllegalStateException(e);
+ }
+ return activeCentroids;
+ }
+
+ /** Extracts the indexed vector value from the row state, or null if absent. */
+ private ImmutableBytesWritable getIndexedVector(Put dataRowState) {
+ if (dataRowState == null) {
+ return null;
+ }
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ ValueGetter valueGetter = new IndexUtil.SimpleValueGetter(dataRowState);
+ if (
+ !indexedExpressions.get(0)
+ .evaluate(new ValueGetterTuple(valueGetter, HConstants.LATEST_TIMESTAMP), ptr)
+ || ptr.getLength() == 0
+ ) {
+ return null;
+ }
+ return ptr;
+ }
+
+ /**
+ * Determines whether the indexed vector value is identical across current and next row states,
+ * indicating no index row key relocation is required.
+ */
+ public boolean isVectorUnchanged(Put currentDataRowState, Put nextDataRowState) {
+ if (!isVectorIndex()) {
+ return false;
+ }
+ ImmutableBytesWritable current = getIndexedVector(currentDataRowState);
+ ImmutableBytesWritable next = getIndexedVector(nextDataRowState);
+ return current != null && next != null && Bytes.equals(current.get(), current.getOffset(),
+ current.getLength(), next.get(), next.getOffset(), next.getLength());
+ }
+
+ /** Returns true if the column is the centroid ID column in the row key of a vector index. */
+ private static boolean isCentroidColumn(PTable index, PColumn column) {
+ return index.isVectorIndex()
+ && MetaDataUtil.VECTOR_CENTROID_ID_COLUMN_NAME.equals(column.getName().getString());
+ }
+
public static class UDFParseNodeVisitor extends StatelessTraverseAllParseNodeVisitor {
private Map<String, UDFParseNode> udfParseNodes;
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 d76f0463e8..86eaab05b8 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
@@ -44,6 +44,7 @@ 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.query.QueryConstants;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.schema.types.PVectorFloat;
@@ -98,6 +99,17 @@ public final class CentroidManager {
return current == null ? now : Math.max(current + 1, now);
}
+ /**
+ * Returns the index name under which the centroids of a vector index are recorded. A view with a
+ * WHERE clause inherits each index of its parent with the name {@code <view>#<index>}. The
+ * inherited index uses the centroids of the parent index, so this method removes all view
+ * prefixes.
+ */
+ public static String getCentroidIndexName(String indexName) {
+ return indexName
+ .substring(indexName.lastIndexOf(QueryConstants.CHILD_VIEW_INDEX_NAME_SEPARATOR) + 1);
+ }
+
/** Writes the centroid vectors of a generation with IDs from 0, and commits the connection. */
public static void persistCentroids(Connection conn, String indexName, long generation,
List<float[]> centroids) throws SQLException {
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 26ce747202..eba4c72a8f 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
@@ -2007,14 +2007,19 @@ public class MetaDataClient {
return new MutationState(0, 0, connection);
}
- // Vector indexes remain in BUILDING state until centroids are initialized and assigned.
+ // A vector index must have a trained centroid generation before the build. buildVectorIndex
+ // trains the first generation and then builds the index rows. If another rebuild holds the
+ // claim, or the training does not complete, it does not build the index rows.
+ // Indexes remain in BUILDING state if the data table has insufficient training vectors.
if (table.isVectorIndex()) {
- MutationState state = trainVectorIndex(table, dataTable);
- if (ValidateLastDDLTimestampUtil.getValidateLastDdlTimestampEnabled(connection)) {
- connection.removeTable(connection.getTenantId(), dataTable.getName().getString(), null,
- dataTable.getTimeStamp());
+ table = trainVectorIndex(table, dataTable);
+ if (table.getVectorCentroidGeneration() == null) {
+ if (ValidateLastDDLTimestampUtil.getValidateLastDdlTimestampEnabled(connection)) {
+ connection.removeTable(connection.getTenantId(), dataTable.getName().getString(), null,
+ dataTable.getTimeStamp());
+ }
+ return new MutationState(0, 0, connection);
}
- return state;
}
// If our connection is at a fixed point-in-time, we need to open a new
@@ -2034,17 +2039,17 @@ public class MetaDataClient {
}
/**
- * Trains and records the initial centroid generation for a vector index, retaining BUILDING
- * state. Training is deferred if the data table contains fewer non-null vectors than requested
- * lists.
+ * 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.
+ * @return refreshed index metadata from catalog
*/
- private MutationState trainVectorIndex(PTable index, PTable dataTable) throws SQLException {
+ private PTable trainVectorIndex(PTable index, PTable dataTable) throws SQLException {
try (PhoenixConnection internal = CentroidManager.newInternalConnection(connection)) {
VectorIndexTrainer.trainAndRecord(internal, dataTable, index);
}
connection.removeTable(connection.getTenantId(), index.getName().getString(),
dataTable.getName().getString(), HConstants.LATEST_TIMESTAMP);
- return new MutationState(0, 0, connection);
+ return connection.getTableNoCache(index.getName().getString());
}
public MutationState createCDC(CreateCDCStatement statement) throws SQLException {
@@ -6751,6 +6756,13 @@ public class MetaDataClient {
SQLExceptionCode.CANNOT_ALTER_TO_BE_TXN_IF_TXNS_DISABLED).setSchemaName(schemaName)
.setTableName(tableName).build().buildException();
}
+ // A table with a vector index cannot become transactional
+ for (PTable index : table.getIndexes()) {
+ if (index.isVectorIndex()) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.VECTOR_INDEX_ON_TRANSACTIONAL_TABLE)
+ .setSchemaName(schemaName).setTableName(tableName).build().buildException();
+ }
+ }
// cannot make a table transactional if it has a row timestamp column
if (SchemaUtil.hasRowTimestampColumn(table)) {
throw new SQLExceptionInfo.Builder(
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/IndexUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/IndexUtil.java
index 7330e537bf..c62a51c5f0 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/IndexUtil.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/IndexUtil.java
@@ -347,6 +347,7 @@ public class IndexUtil {
try {
final ImmutableBytesPtr ptr = new ImmutableBytesPtr();
IndexMaintainer maintainer = index.getIndexMaintainer(table, connection);
+ maintainer.loadCentroids(connection);
List<Mutation> indexMutations = Lists.newArrayListWithExpectedSize(dataMutations.size());
for (final Mutation dataMutation : dataMutations) {
long ts = MetaDataUtil.getClientTimeStamp(dataMutation);
@@ -403,8 +404,12 @@ public class IndexUtil {
regionStartKey = tableRegionLocation.getRegion().getStartKey();
regionEndkey = tableRegionLocation.getRegion().getEndKey();
}
- indexMutations.add(maintainer.buildUpdateMutation(kvBuilder, valueGetter, ptr, ts,
- regionStartKey, regionEndkey, false));
+ Put indexPut = maintainer.buildUpdateMutation(kvBuilder, valueGetter, ptr, ts,
+ regionStartKey, regionEndkey, false);
+ // The mutation is null if the row has no index row, for example if its vector is null
+ if (indexPut != null) {
+ indexMutations.add(indexPut);
+ }
}
}
return indexMutations;
@@ -848,12 +853,14 @@ public class IndexUtil {
}
public static boolean isCoveredGlobalIndex(final PTable table) {
- return table.getIndexType() == PTable.IndexType.GLOBAL;
+ return table.getIndexType() == PTable.IndexType.GLOBAL
+ || table.getIndexType() == PTable.IndexType.VECTOR_GLOBAL;
}
public static boolean isGlobalIndex(final PTable table) {
return table.getIndexType() == PTable.IndexType.GLOBAL
- || table.getIndexType() == PTable.IndexType.UNCOVERED_GLOBAL;
+ || table.getIndexType() == PTable.IndexType.UNCOVERED_GLOBAL
+ || table.getIndexType() == PTable.IndexType.VECTOR_GLOBAL;
}
public static boolean shouldIndexBeUsedForUncoveredQuery(final TableRef tableRef) {
diff --git a/phoenix-core-client/src/main/protobuf/ServerCachingService.proto b/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
index 2c4f358ccc..58911b6833 100644
--- a/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
+++ b/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
@@ -74,6 +74,9 @@ message IndexMaintainer {
optional bool isCDCIndex = 31;
optional int32 dataTableSaltBuckets = 32;
optional string indexConsistency = 33;
+ optional string vectorAlgorithm = 34;
+ optional string distanceMetric = 35;
+ optional int64 centroidGeneration = 36;
}
message TransformMaintainer {
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 5125c783e4..afa016154a 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
@@ -84,6 +84,7 @@ import org.apache.hadoop.io.WritableUtils;
import org.apache.htrace.Span;
import org.apache.htrace.Trace;
import org.apache.htrace.TraceScope;
+import org.apache.phoenix.cache.VectorCentroidCache;
import org.apache.phoenix.compile.ScanRanges;
import org.apache.phoenix.coprocessor.DelegateRegionCoprocessorEnvironment;
import org.apache.phoenix.coprocessor.ServerScanUtil;
@@ -549,6 +550,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
}
this.builder = new IndexBuildManager(env);
+ VectorCentroidCache.setServerConfiguration(env.getConfiguration());
// Clone the config since it is shared
DelegateRegionCoprocessorEnvironment indexWriterEnv =
new DelegateRegionCoprocessorEnvironment(env, ConnectionType.INDEX_WRITER_CONNECTION);
@@ -1416,14 +1418,18 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
indexUpdates.put(hTableInterfaceReference, deleteColumn);
}
}
- // Delete the current index row if the new index key is different from the
- // current one and the index is not a CDC index
- if (currentDataRowState != null) {
+ // Delete the current index row if the new index key differs from the current one
+ // and the index is not CDC. Skip centroid assignment when the indexed vector is unchanged.
+ if (
+ currentDataRowState != null
+ && !indexMaintainer.isVectorUnchanged(currentDataRowState, nextDataRowState)
+ ) {
ValueGetter currentDataRowVG = new IndexUtil.SimpleValueGetter(currentDataRowState);
+ // Null if the current row has no index row, for example if its vector is null
byte[] indexRowKeyForCurrentDataRow = indexMaintainer.buildRowKey(currentDataRowVG,
rowKeyPtr, null, null, ts, encodedRegionName);
if (
- !indexMaintainer.isCDCIndex()
+ indexRowKeyForCurrentDataRow != null && !indexMaintainer.isCDCIndex()
&& Bytes.compareTo(indexPut.getRow(), indexRowKeyForCurrentDataRow) != 0
) {
Mutation del = indexMaintainer.buildRowDeleteMutation(indexRowKeyForCurrentDataRow,
@@ -1987,6 +1993,16 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
ServerIndexUtil.setDeleteAttributes(miniBatchOp);
}
+ // Load the centroids of each vector index before the row locks. A cache miss reads
+ // SYSTEM.VECTOR_CENTROID, and the batch must not hold row locks during that read.
+ for (IndexMaintainer indexMaintainer : indexMetaData.getIndexMaintainers()) {
+ try {
+ indexMaintainer.loadCentroids(null);
+ } catch (SQLException e) {
+ throw ClientUtil.createIOException("Unable to load vector index centroids", e);
+ }
+ }
+
// Exclusively lock all rows to do consistent writes over multiple tables
// (i.e., the data and its index tables)
populateRowsToLock(miniBatchOp, context);
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
index 8662dbdec2..9530227534 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
@@ -71,6 +71,8 @@ import org.apache.phoenix.hbase.index.ValueGetter;
import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.index.vector.CentroidManager;
+import org.apache.phoenix.index.vector.VectorIndexTrainer;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixResultSet;
import org.apache.phoenix.mapreduce.CsvBulkImportUtil;
@@ -846,6 +848,10 @@ public class IndexTool extends Configured implements Tool {
createIndexToolTables(conn);
if (dataTable != null && indexTable != null) {
setupIndexAndDataTable(conn);
+ if (pIndexTable.isVectorIndex() && pIndexTable.getVectorCentroidGeneration() == null) {
+ LOGGER.info("Vector index {} has no trained centroids, leaving it unbuilt", qIndexTable);
+ return 0;
+ }
checkIfFeatureApplicable(startTime, endTime, lastVerifyTime, pDataTable, isLocalIndexBuild);
if (shouldDeleteBeforeRebuild) {
deleteBeforeRebuild(conn);
@@ -1002,6 +1008,18 @@ public class IndexTool extends Configured implements Tool {
.replace(QueryConstants.NAME_SEPARATOR, QueryConstants.NAMESPACE_SEPARATOR));
}
indexType = pIndexTable.getIndexType();
+ if (
+ pIndexTable.isVectorIndex() && pIndexTable.getVectorCentroidGeneration() == null
+ && sourceTable == SourceTable.DATA_TABLE_SOURCE && indexVerifyType != IndexVerifyType.ONLY
+ ) {
+ // Train initial centroid generation for ASYNC vector indexes prior to build. A run that only
+ // verifies builds no index rows and must not train.
+ PhoenixConnection pconn = connection.unwrap(PhoenixConnection.class);
+ try (PhoenixConnection internal = CentroidManager.newInternalConnection(pconn)) {
+ VectorIndexTrainer.trainAndRecord(internal, pDataTable, pIndexTable);
+ }
+ pIndexTable = pconn.getTableNoCache(pIndexTable.getName().getString());
+ }
qIndexTable = SchemaUtil.getQualifiedTableName(schemaName, indexTable);
if (SchemaUtil.isNamespaceMappingEnabled(PTableType.SYSTEM, getConf())) {
qIndexTable =
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexIT.java
index 05a11f4ac2..914be06ea8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexIT.java
@@ -37,6 +37,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_DISTANCE_ME
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_INDEX_ALGORITHM_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_IVF_LISTS_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VECTOR_IVF_SAMPLE_SIZE_BYTES;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -54,28 +55,42 @@ import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
+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.client.TableDescriptor;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils.BlockingRpcCallback;
import org.apache.hadoop.hbase.ipc.ServerRpcController;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.mapreduce.Counters;
import org.apache.phoenix.coprocessor.generated.MetaDataProtos;
import org.apache.phoenix.coprocessor.generated.MetaDataProtos.CreateTableRequest;
import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataResponse;
import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataService;
import org.apache.phoenix.coprocessorclient.MetaDataProtocol;
+import org.apache.phoenix.end2end.index.IndexTestUtil;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.hbase.index.util.VersionUtil;
+import org.apache.phoenix.index.vector.CentroidManager;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.mapreduce.index.IndexTool;
+import org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters;
import org.apache.phoenix.protobuf.ProtobufUtil;
+import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PIndexState;
@@ -83,6 +98,7 @@ import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableImpl;
import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.schema.types.PBson;
import org.apache.phoenix.schema.types.PDataType;
@@ -90,10 +106,12 @@ import org.apache.phoenix.schema.types.PDouble;
import org.apache.phoenix.schema.types.PInteger;
import org.apache.phoenix.schema.types.PLong;
import org.apache.phoenix.schema.types.PVarchar;
+import org.apache.phoenix.schema.types.PVectorDouble;
import org.apache.phoenix.schema.types.PVectorFloat;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.ClientUtil;
import org.apache.phoenix.util.Closeables;
+import org.apache.phoenix.util.EncodedColumnsUtil;
import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PropertiesUtil;
@@ -619,6 +637,13 @@ public class VectorIndexIT extends ParallelStatsDisabledIT {
assertEquals(MetaDataUtil.VECTOR_CENTROID_ID_COLUMN_NAME,
pkColumns.get(0).getName().getString());
assertEquals(":ID", pkColumns.get(1).getName().getString());
+
+ // The index keeps the indexed vector column in the data family with the PVectorDouble type
+ PColumn vectorCol = indexTable.getColumnForColumnName("0:V");
+ assertNotNull("Vector column 0:V must exist in index table", vectorCol);
+ assertEquals("Double vector column must have PVectorDouble data type", PVectorDouble.INSTANCE,
+ vectorCol.getDataType());
+ assertEquals(Integer.valueOf(64), vectorCol.getMaxLength());
}
}
@@ -916,12 +941,37 @@ public class VectorIndexIT extends ParallelStatsDisabledIT {
@Test
public void testDropVectorIndexIfExists() throws Exception {
String tableName = "T_IF_EXISTS_" + generateUniqueName();
- String indexName = "IDX_NON_EXISTENT_" + generateUniqueName();
+ String indexName = "IDX_IF_EXISTS_" + generateUniqueName();
- try (Connection conn = DriverManager.getConnection(getUrl())) {
+ try (Connection conn = getDropMetadataConnection()) {
try (Statement stmt = conn.createStatement()) {
stmt.execute(
"CREATE TABLE " + tableName + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 32))");
+ stmt.execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) "
+ + "WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)");
+ }
+
+ // Make sure that SYSTEM.CATALOG has the index before the drop
+ try (Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery("SELECT 1 FROM SYSTEM.CATALOG WHERE TABLE_NAME = '"
+ + indexName + "' AND TABLE_SCHEM IS NULL AND TENANT_ID IS NULL")) {
+ assertTrue("Index must exist in SYSTEM.CATALOG before drop", rs.next());
+ }
+
+ // DROP INDEX IF EXISTS removes the index if it exists
+ try (Statement stmt = conn.createStatement()) {
+ stmt.execute("DROP INDEX IF EXISTS " + indexName + " ON " + tableName);
+ }
+
+ try (Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery("SELECT 1 FROM SYSTEM.CATALOG WHERE TABLE_NAME = '"
+ + indexName + "' AND TABLE_SCHEM IS NULL AND TENANT_ID IS NULL")) {
+ assertFalse("Index must be removed from SYSTEM.CATALOG after DROP INDEX IF EXISTS",
+ rs.next());
+ }
+
+ // DROP INDEX IF EXISTS must not fail if the index does not exist
+ try (Statement stmt = conn.createStatement()) {
stmt.execute("DROP INDEX IF EXISTS " + indexName + " ON " + tableName);
}
}
@@ -961,11 +1011,11 @@ public class VectorIndexIT extends ParallelStatsDisabledIT {
}
/**
- * Verifies that synchronous index creation trains initial centroids, records active generation
- * metadata, and retains BUILDING state pending population.
+ * Verifies synchronous vector index population, building verified index rows keyed by centroid ID
+ * and transitioning index state to ACTIVE.
*/
@Test
- public void testSynchronousVectorIndexTraining() throws Exception {
+ public void testSynchronousVectorIndexPopulationAndActivation() throws Exception {
String tableName = "T_VEC_POP_" + generateUniqueName();
String indexName = "IDX_VEC_POP_" + generateUniqueName();
try (Connection conn = DriverManager.getConnection(getUrl())) {
@@ -980,14 +1030,29 @@ public class VectorIndexIT extends ParallelStatsDisabledIT {
"CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) " + "INCLUDE (LABEL) "
+ "WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)");
}
- PTable index = conn.unwrap(PhoenixConnection.class).getTableNoCache(indexName);
- assertEquals(PIndexState.BUILDING, index.getIndexState());
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable index = pconn.getTableNoCache(indexName);
+ assertEquals(PIndexState.ACTIVE, index.getIndexState());
Long generation = index.getVectorCentroidGeneration();
assertNotNull("Training records a generation", generation);
assertTrue("The first generation is the training time", generation >= before);
- assertTrue(index.getVectorIvfLists() >= 4);
- assertEquals(index.getVectorIvfLists().intValue(),
- countCentroids(conn, indexName, generation));
+ int lists = index.getVectorIvfLists();
+ assertTrue(lists >= 4);
+ assertEquals(lists, countCentroids(conn, indexName, generation));
+
+ String sql = "SELECT \"_CENTROID_ID\", \":ID\", \"0:LABEL\" FROM " + indexName;
+ int rows = 0;
+ try (ResultSet rs = conn.createStatement().executeQuery(sql)) {
+ while (rs.next()) {
+ rows++;
+ assertTrue(rs.getInt(1) >= 0 && rs.getInt(1) < lists);
+ assertNotNull(rs.getString(2));
+ assertNotNull(rs.getString(3));
+ }
+ }
+ assertEquals(100, rows);
+ IndexTestUtil.assertRowsForEmptyColValue(conn, indexName, QueryConstants.VERIFIED_BYTES);
+ assertIndexVerifies(tableName, indexName, 100);
}
}
@@ -1082,4 +1147,805 @@ public class VectorIndexIT extends ParallelStatsDisabledIT {
assertTrue("Sampled " + rows + " of 1000 at p=0.3", rows > 200 && rows < 400);
}
}
+
+ private static final List<float[]> KNOWN_CENTROIDS =
+ Arrays.asList(new float[] { 0.0f, 0.0f, 0.0f, 1.0f }, // ID 0
+ new float[] { 0.0f, 0.0f, 1.0f, 0.0f }, // ID 1
+ new float[] { 1.0f, 0.0f, 0.0f, 0.0f }, // ID 2
+ new float[] { 0.0f, 1.0f, 0.0f, 0.0f }); // ID 3
+
+ /**
+ * Initializes test table and active vector index configured with deterministic centroid
+ * positions.
+ */
+ private void setupTableAndKnownCentroids(Connection conn, String tableName, String indexName)
+ throws Exception {
+ setupTableAndKnownCentroids(conn, tableName, indexName, "FLOAT", "");
+ }
+
+ private void setupTableAndKnownCentroids(Connection conn, String tableName, String indexName,
+ String elementType, String tableOptions) throws Exception {
+ try (Statement stmt = conn.createStatement()) {
+ stmt.execute("CREATE TABLE " + tableName + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR("
+ + elementType + ", 4), LABEL VARCHAR) " + tableOptions);
+ stmt.execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) "
+ + "INCLUDE (LABEL) WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)");
+ }
+ recordKnownCentroids(conn, indexName, KNOWN_CENTROIDS);
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ IndexUtil.updateIndexState(pconn, indexName, PIndexState.ACTIVE, 0L);
+ pconn.removeTable(pconn.getTenantId(), indexName, null, HConstants.LATEST_TIMESTAMP);
+ pconn.removeTable(pconn.getTenantId(), tableName, null, HConstants.LATEST_TIMESTAMP);
+ }
+
+ /** Persists specified centroids as a new generation for the given index. */
+ private static long recordKnownCentroids(Connection conn, String indexName,
+ List<float[]> centroids) throws SQLException {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ try (PhoenixConnection internal = CentroidManager.newInternalConnection(pconn)) {
+ PTable index = internal.getTableNoCache(indexName);
+ long generation = CentroidManager.nextGeneration(index.getVectorCentroidGeneration());
+ CentroidManager.persistCentroids(internal, indexName, generation, centroids);
+ CentroidManager.setGenerationAndLists(internal, index, generation, centroids.size());
+ return generation;
+ }
+ }
+
+ private List<byte[]> getHBaseRowKeys(PhoenixConnection pconn, PTable table) throws Exception {
+ byte[] physicalNameBytes = table.getPhysicalName().getBytes();
+ List<byte[]> rowKeys = new ArrayList<>();
+ try (Table hTable = pconn.getQueryServices().getTable(physicalNameBytes);
+ org.apache.hadoop.hbase.client.ResultScanner scanner =
+ hTable.getScanner(new org.apache.hadoop.hbase.client.Scan())) {
+ for (org.apache.hadoop.hbase.client.Result r : scanner) {
+ rowKeys.add(r.getRow());
+ }
+ }
+ return rowKeys;
+ }
+
+ private int extractCentroidId(byte[] rowKey) {
+ return (Integer) PInteger.INSTANCE.toObject(rowKey, 0, Bytes.SIZEOF_INT, PInteger.INSTANCE,
+ SortOrder.getDefault());
+ }
+
+ @Test
+ public void testVectorInsertGeneratesIndexRow() throws Exception {
+ String tableName = "T_VEC_INS_" + generateUniqueName();
+ String indexName = "IDX_VEC_INS_" + generateUniqueName();
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName);
+
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable indexTable = pconn.getTableNoCache(indexName);
+
+ // Upsert test vector mapping to centroid ID 2
+ String upsertSql = "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_1");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "lbl_1");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ // Verify physical HBase index row key contains assigned centroid ID
+ List<byte[]> rowKeys = getHBaseRowKeys(pconn, indexTable);
+ assertEquals("Expected exactly 1 index row", 1, rowKeys.size());
+ assertEquals("Centroid prefix must be 2", 2, extractCentroidId(rowKeys.get(0)));
+
+ // Verify index scan via SQL
+ String selectSql = "SELECT \"_CENTROID_ID\", \":ID\", \"0:LABEL\" FROM " + indexName;
+ try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(selectSql)) {
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+ assertEquals("row_1", rs.getString(2));
+ assertEquals("lbl_1", rs.getString(3));
+ assertFalse(rs.next());
+ }
+ }
+ }
+
+ @Test
+ public void testVectorUpdateWithCentroidChange() throws Exception {
+ String tableName = "T_VEC_UPD_" + generateUniqueName();
+ String indexName = "IDX_VEC_UPD_" + generateUniqueName();
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName);
+
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable indexTable = pconn.getTableNoCache(indexName);
+
+ // Insert initial row mapping to centroid 2
+ String upsertSql = "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_1");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "lbl_1");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ List<byte[]> rowKeysBefore = getHBaseRowKeys(pconn, indexTable);
+ assertEquals(1, rowKeysBefore.size());
+ assertEquals(2, extractCentroidId(rowKeysBefore.get(0)));
+
+ // Update vector to value mapping to centroid 0
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_1");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 0.0f, 0.0f, 0.0f, 1.0f }));
+ ps.setString(3, "lbl_updated");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ // Verify index row key relocation from old centroid to new centroid
+ List<byte[]> rowKeysAfter = getHBaseRowKeys(pconn, indexTable);
+ assertEquals("Expected exactly 1 index row after vector update", 1, rowKeysAfter.size());
+ assertEquals("Centroid prefix must be updated to 0", 0,
+ extractCentroidId(rowKeysAfter.get(0)));
+
+ // Verify updated row visibility via SQL scan
+ String selectSql = "SELECT \"_CENTROID_ID\", \":ID\", \"0:LABEL\" FROM " + indexName;
+ try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(selectSql)) {
+ assertTrue(rs.next());
+ assertEquals(0, rs.getInt(1));
+ assertEquals("row_1", rs.getString(2));
+ assertEquals("lbl_updated", rs.getString(3));
+ assertFalse(rs.next());
+ }
+ }
+ }
+
+ @Test
+ public void testVectorCoveredColumnUpdateWithoutVectorChange() throws Exception {
+ String tableName = "T_VEC_COV_" + generateUniqueName();
+ String indexName = "IDX_VEC_COV_" + generateUniqueName();
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName);
+
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable indexTable = pconn.getTableNoCache(indexName);
+
+ // Insert initial row mapping to centroid 2
+ String upsertSql = "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_1");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "initial_label");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ List<byte[]> rowKeysBefore = getHBaseRowKeys(pconn, indexTable);
+ assertEquals(1, rowKeysBefore.size());
+ assertEquals(2, extractCentroidId(rowKeysBefore.get(0)));
+
+ // Update covered column without modifying vector value
+ try (PreparedStatement ps =
+ conn.prepareStatement("UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)")) {
+ ps.setString(1, "row_1");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "updated_label");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ // Verify index row key and centroid prefix remain unchanged
+ List<byte[]> rowKeysAfter = getHBaseRowKeys(pconn, indexTable);
+ assertEquals("Expected exactly 1 index row", 1, rowKeysAfter.size());
+ assertEquals("Centroid prefix must still be 2", 2, extractCentroidId(rowKeysAfter.get(0)));
+
+ // Verify in-place update of covered column
+ String selectSql = "SELECT \"_CENTROID_ID\", \":ID\", \"0:LABEL\" FROM " + indexName;
+ try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(selectSql)) {
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+ assertEquals("row_1", rs.getString(2));
+ assertEquals("updated_label", rs.getString(3));
+ assertFalse(rs.next());
+ }
+ }
+ }
+
+ @Test
+ public void testVectorDeleteRemovesIndexRow() throws Exception {
+ String tableName = "T_VEC_DEL_" + generateUniqueName();
+ String indexName = "IDX_VEC_DEL_" + generateUniqueName();
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName);
+
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable indexTable = pconn.getTableNoCache(indexName);
+
+ // Insert initial row mapping to centroid 2
+ String upsertSql = "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_1");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "lbl_1");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ List<byte[]> rowKeysBefore = getHBaseRowKeys(pconn, indexTable);
+ assertEquals(1, rowKeysBefore.size());
+
+ // Delete base table row
+ try (Statement stmt = conn.createStatement()) {
+ stmt.execute("DELETE FROM " + tableName + " WHERE ID = 'row_1'");
+ }
+ conn.commit();
+
+ // Verify physical index row deletion
+ List<byte[]> rowKeysAfter = getHBaseRowKeys(pconn, indexTable);
+ assertEquals("Expected 0 index rows after delete", 0, rowKeysAfter.size());
+
+ // Verify index table scan returns zero rows
+ String selectSql = "SELECT COUNT(*) FROM " + indexName;
+ try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(selectSql)) {
+ assertTrue(rs.next());
+ assertEquals(0, rs.getInt(1));
+ }
+ }
+ }
+
+ @Test
+ public void testNullVectorExcludedFromIndex() throws Exception {
+ String tableName = "T_VEC_NULL_" + generateUniqueName();
+ String indexName = "IDX_VEC_NULL_" + generateUniqueName();
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName);
+
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable indexTable = pconn.getTableNoCache(indexName);
+
+ // Insert row with null vector
+ String upsertSql = "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_null");
+ ps.setNull(2, java.sql.Types.ARRAY);
+ ps.setString(3, "null_label");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ // Verify null vector does not generate index row
+ List<byte[]> rowKeys = getHBaseRowKeys(pconn, indexTable);
+ assertEquals("Null vector must not produce an index row", 0, rowKeys.size());
+
+ // Verify base table row presence
+ try (Statement stmt = conn.createStatement(); ResultSet rs =
+ stmt.executeQuery("SELECT ID, LABEL FROM " + tableName + " WHERE ID = 'row_null'")) {
+ assertTrue("Base table must contain the null vector row", rs.next());
+ assertEquals("row_null", rs.getString(1));
+ assertEquals("null_label", rs.getString(2));
+ }
+ }
+ }
+
+ @Test
+ public void testVectorUnchangedUpdateMaintainsIndexRowAndCoveredColumns() throws Exception {
+ String tableName = "T_VEC_ZUPD_" + generateUniqueName();
+ String indexName = "IDX_VEC_ZUPD_" + generateUniqueName();
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName);
+
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable indexTable = pconn.getTableNoCache(indexName);
+
+ // Insert initial row
+ String upsertSql = "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_alloc_u");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "lbl_initial");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ // Verify initial index row key and centroid prefix
+ List<byte[]> rowKeysBefore = getHBaseRowKeys(pconn, indexTable);
+ assertEquals(1, rowKeysBefore.size());
+ assertEquals(2, extractCentroidId(rowKeysBefore.get(0)));
+
+ // Partial update modifying covered column only
+ try (Statement stmt = conn.createStatement()) {
+ stmt.execute(
+ "UPSERT INTO " + tableName + " (ID, LABEL) VALUES ('row_alloc_u', 'lbl_updated')");
+ }
+ conn.commit();
+
+ // Verify index row key remains unchanged
+ List<byte[]> rowKeysAfterPartial = getHBaseRowKeys(pconn, indexTable);
+ assertEquals("Expected exactly 1 index row after partial update", 1,
+ rowKeysAfterPartial.size());
+ assertEquals(2, extractCentroidId(rowKeysAfterPartial.get(0)));
+
+ // Verify vector column payload is preserved during partial update
+ PColumn indexVecCol = indexTable.getColumnForColumnName("0:V");
+ try (
+ Table hTable = pconn.getQueryServices().getTable(indexTable.getPhysicalName().getBytes())) {
+ Result r = hTable.get(new Get(rowKeysAfterPartial.get(0)));
+ byte[] storedVec =
+ r.getValue(indexVecCol.getFamilyName().getBytes(), indexVecCol.getColumnQualifierBytes());
+ assertNotNull("index row must still carry the vector after a covered-only update",
+ storedVec);
+ assertArrayEquals(PVectorFloat.INSTANCE.toBytes(new float[] { 1.0f, 0.0f, 0.0f, 0.0f }),
+ storedVec);
+ }
+
+ // Full update supplying identical vector
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_alloc_u");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "lbl_updated_again");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ // Verify index row key remains unchanged after full update with identical vector
+ List<byte[]> rowKeys = getHBaseRowKeys(pconn, indexTable);
+ assertEquals("Expected exactly 1 index row after full unchanged update", 1, rowKeys.size());
+ assertEquals(2, extractCentroidId(rowKeys.get(0)));
+
+ // Verify updated covered column values
+ String selectSql = "SELECT \"_CENTROID_ID\", \":ID\", \"0:LABEL\" FROM " + indexName;
+ try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(selectSql)) {
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+ assertEquals("row_alloc_u", rs.getString(2));
+ assertEquals("lbl_updated_again", rs.getString(3));
+ assertFalse(rs.next());
+ }
+ }
+ }
+
+ @Test
+ public void testVerifiedMarkerPresentAfterCommit() throws Exception {
+ String tableName = "T_VEC_VER_" + generateUniqueName();
+ String indexName = "IDX_VEC_VER_" + generateUniqueName();
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName);
+
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable indexTable = pconn.getTableNoCache(indexName);
+ byte[] physicalIndexName = indexTable.getPhysicalName().getBytes();
+
+ // Insert row mapping to centroid 2
+ String upsertSql = "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, "row_v1");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "lbl_v1");
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ byte[] emptyCF = SchemaUtil.getEmptyColumnFamily(indexTable);
+ byte[] emptyCQ = EncodedColumnsUtil.getEmptyKeyValueInfo(indexTable).getFirst();
+
+ // Verify VERIFIED empty column cell directly on physical HBase table
+ try (Table hTable = pconn.getQueryServices().getTable(physicalIndexName);
+ ResultScanner scanner = hTable.getScanner(new Scan())) {
+ Result result = scanner.next();
+ assertNotNull("Expected index row in HBase table", result);
+
+ byte[] emptyColVal = result.getValue(emptyCF, emptyCQ);
+ assertNotNull("Empty column marker cell must be present on index row", emptyColVal);
+ assertTrue("Empty column marker must be VERIFIED_BYTES",
+ Bytes.equals(QueryConstants.VERIFIED_BYTES, emptyColVal));
+
+ assertNull("Expected exactly 1 index row", scanner.next());
+ }
+
+ // Verify verification status via IndexTestUtil
+ IndexTestUtil.assertRowsForEmptyColValue(conn, indexName, QueryConstants.VERIFIED_BYTES);
+ }
+ }
+
+ /**
+ * Verifies read repair cleans up stale index entries under outdated centroids and rebuilds
+ * verified rows matching the current vector.
+ */
+ @Test
+ public void testReadRepairWithCentroidReassignment() throws Exception {
+ String tableName = "T_VEC_RR_" + generateUniqueName();
+ String indexName = "IDX_VEC_RR_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName);
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+
+ // Populate base table row while index is disabled
+ IndexUtil.updateIndexState(pconn, indexName, PIndexState.DISABLE, 0L);
+ pconn.removeTable(pconn.getTenantId(), tableName, null, HConstants.LATEST_TIMESTAMP);
+ try (PreparedStatement ps =
+ conn.prepareStatement("UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)")) {
+ ps.setString(1, "repair_row_1");
+ // Vector maps to centroid 2
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setString(3, "correct_label");
+ ps.executeUpdate();
+ }
+ conn.commit();
+ IndexUtil.updateIndexState(pconn, indexName, PIndexState.BUILDING, 0L);
+ IndexUtil.updateIndexState(pconn, indexName, PIndexState.ACTIVE, 0L);
+ pconn.removeTable(pconn.getTenantId(), indexName, null, HConstants.LATEST_TIMESTAMP);
+ pconn.removeTable(pconn.getTenantId(), tableName, null, HConstants.LATEST_TIMESTAMP);
+
+ PTable indexTable = pconn.getTableNoCache(indexName);
+ byte[] physicalIndexName = indexTable.getPhysicalName().getBytes();
+ byte[] centroid0RowKey =
+ ByteUtil.concat(PInteger.INSTANCE.toBytes(0), Bytes.toBytes("repair_row_1"));
+ byte[] centroid2RowKey =
+ ByteUtil.concat(PInteger.INSTANCE.toBytes(2), Bytes.toBytes("repair_row_1"));
+ byte[] emptyCF = SchemaUtil.getEmptyColumnFamily(indexTable);
+ byte[] emptyCQ = EncodedColumnsUtil.getEmptyKeyValueInfo(indexTable).getFirst();
+ PColumn labelCol = indexTable.getColumnForColumnName("0:LABEL");
+ byte[] labelCF = labelCol.getFamilyName().getBytes();
+ byte[] labelCQ = labelCol.getColumnQualifierBytes();
+
+ // Simulate partial failure leaving unverified index row under incorrect centroid 0
+ try (Table hIndexTable = pconn.getQueryServices().getTable(physicalIndexName)) {
+ assertTrue(getHBaseRowKeys(pconn, indexTable).isEmpty());
+ Put stalePut = new Put(centroid0RowKey);
+ stalePut.addColumn(emptyCF, emptyCQ, QueryConstants.UNVERIFIED_BYTES);
+ stalePut.addColumn(labelCF, labelCQ, Bytes.toBytes("stale_label"));
+ hIndexTable.put(stalePut);
+ }
+
+ // Read repair triggers during index query
+ String selectSql = "SELECT \"_CENTROID_ID\", \":ID\", \"0:LABEL\" FROM " + indexName;
+ try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(selectSql)) {
+ assertTrue("Query through index should return the repaired row", rs.next());
+ assertEquals("Repaired centroid ID must be 2", 2, rs.getInt(1));
+ assertEquals("repair_row_1", rs.getString(2));
+ assertEquals("correct_label", rs.getString(3));
+ assertFalse("Only one row should be returned", rs.next());
+ }
+
+ try (Table hIndexTable = pconn.getQueryServices().getTable(physicalIndexName)) {
+ Result r2 = hIndexTable.get(new Get(centroid2RowKey));
+ assertFalse("Repaired centroid 2 row must exist after read repair", r2.isEmpty());
+ assertArrayEquals(QueryConstants.VERIFIED_BYTES, r2.getValue(emptyCF, emptyCQ));
+ assertEquals("correct_label", Bytes.toString(r2.getValue(labelCF, labelCQ)));
+ // Stale unverified row remains unserved until background cleanup
+ Result r0 = hIndexTable.get(new Get(centroid0RowKey));
+ assertTrue(r0.isEmpty()
+ || Bytes.equals(QueryConstants.UNVERIFIED_BYTES, r0.getValue(emptyCF, emptyCQ)));
+ }
+ }
+ }
+
+ private int findNearestCentroid(float[] v, List<float[]> centroids) {
+ int bestId = -1;
+ double bestDistSq = Double.MAX_VALUE;
+ for (int c = 0; c < centroids.size(); c++) {
+ float[] centroid = centroids.get(c);
+ double distSq = 0.0;
+ for (int d = 0; d < v.length; d++) {
+ double diff = v[d] - centroid[d];
+ distSq += diff * diff;
+ }
+ if (distSq < bestDistSq) {
+ bestDistSq = distSq;
+ bestId = c;
+ }
+ }
+ return bestId;
+ }
+
+ /** Verifies covered vector columns of differing element widths are preserved in index rows. */
+ @Test
+ public void testCoveredDoubleVectorColumn() throws Exception {
+ String tableName = "T_VEC_COVD_" + generateUniqueName();
+ String indexName = "IDX_VEC_COVD_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ try (Statement stmt = conn.createStatement()) {
+ stmt.execute("CREATE TABLE " + tableName + " (ID VARCHAR NOT NULL PRIMARY KEY, "
+ + "V VECTOR(FLOAT, 4), LABEL VARCHAR, COV_D VECTOR(DOUBLE, 3))");
+ stmt.execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) "
+ + "INCLUDE (LABEL, COV_D) WITH (algorithm = 'IVF', metric = 'L2', lists = 4, "
+ + "sample_size = 100)");
+ }
+ recordKnownCentroids(conn, indexName, KNOWN_CENTROIDS);
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ IndexUtil.updateIndexState(pconn, indexName, PIndexState.ACTIVE, 0L);
+ pconn.removeTable(pconn.getTenantId(), indexName, null, HConstants.LATEST_TIMESTAMP);
+ pconn.removeTable(pconn.getTenantId(), tableName, null, HConstants.LATEST_TIMESTAMP);
+
+ String upsert = "UPSERT INTO " + tableName + " (ID, V, COV_D) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsert)) {
+ ps.setString(1, "r1");
+ ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }));
+ ps.setArray(3, conn.createArrayOf("DOUBLE", new Double[] { 1.5, -2.25, 3.125 }));
+ ps.executeUpdate();
+ }
+ conn.commit();
+ // Update covered vector column only
+ try (PreparedStatement ps =
+ conn.prepareStatement("UPSERT INTO " + tableName + " (ID, COV_D) VALUES (?, ?)")) {
+ ps.setString(1, "r1");
+ ps.setArray(2, conn.createArrayOf("DOUBLE", new Double[] { 4.5, 5.5, -6.5 }));
+ ps.executeUpdate();
+ }
+ conn.commit();
+
+ PTable indexTable = pconn.getTableNoCache(indexName);
+ List<byte[]> rowKeys = getHBaseRowKeys(pconn, indexTable);
+ assertEquals(1, rowKeys.size());
+ assertEquals(2, extractCentroidId(rowKeys.get(0)));
+ PColumn covCol = indexTable.getColumnForColumnName("0:COV_D");
+ PColumn vecCol = indexTable.getColumnForColumnName("0:V");
+ try (
+ Table hTable = pconn.getQueryServices().getTable(indexTable.getPhysicalName().getBytes())) {
+ Result r = hTable.get(new Get(rowKeys.get(0)));
+ assertArrayEquals(PVectorDouble.INSTANCE.toBytes(new double[] { 4.5, 5.5, -6.5 }),
+ r.getValue(covCol.getFamilyName().getBytes(), covCol.getColumnQualifierBytes()));
+ assertArrayEquals(PVectorFloat.INSTANCE.toBytes(new float[] { 1.0f, 0.0f, 0.0f, 0.0f }),
+ r.getValue(vecCol.getFamilyName().getBytes(), vecCol.getColumnQualifierBytes()));
+ }
+ }
+ }
+
+ /** Verifies lifecycle maintenance of double precision indexes across centroid boundaries. */
+ @Test
+ public void testDoubleVectorIndexMaintenance() throws Exception {
+ String tableName = "T_VEC_DBL_" + generateUniqueName();
+ String indexName = "IDX_VEC_DBL_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName, "DOUBLE", "");
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable indexTable = pconn.getTableNoCache(indexName);
+ String upsert = "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsert)) {
+ ps.setString(1, "d1");
+ ps.setArray(2, conn.createArrayOf("DOUBLE", new Double[] { 0.9, 0.1, 0.0, 0.0 }));
+ ps.setString(3, "a");
+ ps.executeUpdate();
+ }
+ conn.commit();
+ List<byte[]> rowKeys = getHBaseRowKeys(pconn, indexTable);
+ assertEquals(1, rowKeys.size());
+ assertEquals(2, extractCentroidId(rowKeys.get(0)));
+
+ try (PreparedStatement ps = conn.prepareStatement(upsert)) {
+ ps.setString(1, "d1");
+ ps.setArray(2, conn.createArrayOf("DOUBLE", new Double[] { 0.0, 0.1, 0.0, 0.9 }));
+ ps.setString(3, "b");
+ ps.executeUpdate();
+ }
+ conn.commit();
+ rowKeys = getHBaseRowKeys(pconn, indexTable);
+ assertEquals(1, rowKeys.size());
+ assertEquals(0, extractCentroidId(rowKeys.get(0)));
+
+ conn.createStatement().execute("DELETE FROM " + tableName + " WHERE ID = 'd1'");
+ conn.commit();
+ assertEquals(0, getHBaseRowKeys(pconn, indexTable).size());
+ }
+ }
+
+ /** Runs IndexTool verification comparing physical index rows against server rebuilt rows. */
+ private static void assertIndexVerifies(String tableName, String indexName, long rows)
+ throws Exception {
+ IndexTool tool = IndexToolIT.runIndexTool(false, null, tableName, indexName, null, 0,
+ IndexTool.IndexVerifyType.ONLY);
+ Counters counters = tool.getJob().getCounters();
+ assertEquals(rows, counters
+ .findCounter(PhoenixIndexToolJobCounters.BEFORE_REBUILD_VALID_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, counters
+ .findCounter(PhoenixIndexToolJobCounters.BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, counters
+ .findCounter(PhoenixIndexToolJobCounters.BEFORE_REBUILD_MISSING_INDEX_ROW_COUNT).getValue());
+ }
+
+ private static float[][] loadRandomVectors(Connection conn, String tableName, String tenantCol,
+ String[] tenants, int rows, long seed) throws SQLException {
+ Random rng = new Random(seed);
+ float[][] vectors = new float[rows][4];
+ String upsert = tenantCol == null
+ ? "UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)"
+ : "UPSERT INTO " + tableName + " (" + tenantCol + ", ID, V, LABEL) VALUES (?, ?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsert)) {
+ for (int i = 0; i < rows; i++) {
+ Float[] vec = new Float[4];
+ for (int d = 0; d < 4; d++) {
+ vec[d] = rng.nextFloat() * 10f;
+ vectors[i][d] = vec[d];
+ }
+ int p = 1;
+ if (tenantCol != null) {
+ ps.setString(p++, tenants[i % tenants.length]);
+ }
+ ps.setString(p++, String.format("row_%03d", i));
+ ps.setArray(p++, conn.createArrayOf("FLOAT", vec));
+ ps.setString(p, "label_" + i);
+ ps.executeUpdate();
+ }
+ }
+ conn.commit();
+ return vectors;
+ }
+
+ /**
+ * Verifies IndexTool execution for ASYNC vector indexes: initial centroid training, population of
+ * verified index rows, and transition to ACTIVE state.
+ */
+ @Test
+ public void testIndexToolTrainsAndBuilds() throws Exception {
+ String tableName = "T_VEC_IT_" + generateUniqueName();
+ String indexName = "IDX_VEC_IT_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ conn.createStatement().execute("CREATE TABLE " + tableName
+ + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 4), LABEL VARCHAR)");
+ float[][] vectors = loadRandomVectors(conn, tableName, null, null, 500, 42);
+ conn.createStatement()
+ .execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) INCLUDE (LABEL)"
+ + " WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100) ASYNC");
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable pIndex = pconn.getTableNoCache(indexName);
+ assertEquals(PIndexState.BUILDING, pIndex.getIndexState());
+ assertNull(pIndex.getVectorCentroidGeneration());
+
+ // Execute IndexTool build
+ IndexToolIT.runIndexTool(false, null, tableName, indexName);
+
+ pIndex = pconn.getTableNoCache(indexName);
+ assertEquals(PIndexState.ACTIVE, pIndex.getIndexState());
+ Long generation = pIndex.getVectorCentroidGeneration();
+ assertNotNull("IndexTool trains the first generation", generation);
+ List<float[]> centroids = CentroidManager.loadCentroids(conn, indexName, generation);
+ assertEquals(pIndex.getVectorIvfLists().intValue(), centroids.size());
+
+ byte[] emptyCF = SchemaUtil.getEmptyColumnFamily(pIndex);
+ byte[] emptyCQ = EncodedColumnsUtil.getEmptyKeyValueInfo(pIndex).getFirst();
+ int rows = 0;
+ try (
+ Table hIndexTable = pconn.getQueryServices().getTable(pIndex.getPhysicalName().getBytes());
+ ResultScanner scanner = hIndexTable.getScanner(new Scan())) {
+ for (Result r : scanner) {
+ rows++;
+ byte[] rowKey = r.getRow();
+ String id = (String) PVarchar.INSTANCE.toObject(rowKey, Bytes.SIZEOF_INT,
+ rowKey.length - Bytes.SIZEOF_INT);
+ int rowIdx = Integer.parseInt(id.replace("row_", ""));
+ assertEquals("Centroid of " + id, findNearestCentroid(vectors[rowIdx], centroids),
+ extractCentroidId(rowKey));
+ assertArrayEquals(QueryConstants.VERIFIED_BYTES, r.getValue(emptyCF, emptyCQ));
+ }
+ }
+ assertEquals(500, rows);
+ assertIndexVerifies(tableName, indexName, 500);
+ }
+ }
+
+ /** Verifies IndexTool defers centroid training and index build when data is insufficient. */
+ @Test
+ public void testIndexToolDefersTrainingOnTooFewVectors() throws Exception {
+ String tableName = "T_VEC_IT_" + generateUniqueName();
+ String indexName = "IDX_VEC_IT_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ conn.createStatement().execute("CREATE TABLE " + tableName
+ + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 4), LABEL VARCHAR)");
+ loadRandomVectors(conn, tableName, null, null, 2, 7);
+ conn.createStatement().execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName
+ + " (V)" + " WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100) ASYNC");
+ IndexTool tool = new IndexTool();
+ tool.setConf(new Configuration(getUtility().getConfiguration()));
+ assertEquals(0, tool.run(new String[] { "-dt", tableName, "-it", indexName, "-runfg" }));
+ PTable pIndex = conn.unwrap(PhoenixConnection.class).getTableNoCache(indexName);
+ assertEquals(PIndexState.BUILDING, pIndex.getIndexState());
+ assertNull(pIndex.getVectorCentroidGeneration());
+ }
+ }
+
+ /** Verifies IndexTool build of multi-tenant vector indexes with composite row keys. */
+ @Test
+ public void testIndexToolBuildsMultiTenantVectorIndex() throws Exception {
+ String tableName = "T_VEC_MT_" + generateUniqueName();
+ String indexName = "IDX_VEC_MT_" + generateUniqueName();
+ String[] tenants = { "TA", "TB" };
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ conn.createStatement()
+ .execute("CREATE TABLE " + tableName + " (TENANT_ID VARCHAR NOT NULL, ID VARCHAR NOT NULL,"
+ + " V VECTOR(FLOAT, 4), LABEL VARCHAR CONSTRAINT PK PRIMARY KEY (TENANT_ID, ID))"
+ + " MULTI_TENANT = true");
+ float[][] vectors = loadRandomVectors(conn, tableName, "TENANT_ID", tenants, 200, 11);
+ conn.createStatement()
+ .execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) INCLUDE (LABEL)"
+ + " WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100) ASYNC");
+
+ IndexToolIT.runIndexTool(false, null, tableName, indexName);
+
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable pIndex = pconn.getTableNoCache(indexName);
+ assertEquals(PIndexState.ACTIVE, pIndex.getIndexState());
+ List<float[]> centroids =
+ CentroidManager.loadCentroids(conn, indexName, pIndex.getVectorCentroidGeneration());
+ int rows = 0;
+ try (
+ Table hIndexTable = pconn.getQueryServices().getTable(pIndex.getPhysicalName().getBytes());
+ ResultScanner scanner = hIndexTable.getScanner(new Scan())) {
+ for (Result r : scanner) {
+ rows++;
+ byte[] rowKey = r.getRow();
+ // Row key format: [TENANT_ID][separator][centroid id][ID]
+ int sep = Bytes.indexOf(rowKey, QueryConstants.SEPARATOR_BYTE);
+ String tenant = Bytes.toString(rowKey, 0, sep);
+ int centroid = (Integer) PInteger.INSTANCE.toObject(rowKey, sep + 1, Bytes.SIZEOF_INT);
+ String id = Bytes.toString(rowKey, sep + 1 + Bytes.SIZEOF_INT,
+ rowKey.length - sep - 1 - Bytes.SIZEOF_INT);
+ int rowIdx = Integer.parseInt(id.replace("row_", ""));
+ assertEquals(tenants[rowIdx % tenants.length], tenant);
+ assertEquals("Centroid of " + id, findNearestCentroid(vectors[rowIdx], centroids),
+ centroid);
+ }
+ }
+ assertEquals(200, rows);
+ assertIndexVerifies(tableName, indexName, 200);
+ }
+ }
+
+ /** Verifies client side index maintenance on immutable tables matches server side row keys. */
+ @Test
+ public void testImmutableTableClientMaintenanceMatchesServerBuild() throws Exception {
+ String tableName = "T_VEC_IMM_" + generateUniqueName();
+ String indexName = "IDX_VEC_IMM_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ setupTableAndKnownCentroids(conn, tableName, indexName, "FLOAT", "IMMUTABLE_ROWS = true");
+ float[][] vectors = loadRandomVectors(conn, tableName, null, null, 50, 3);
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable pIndex = pconn.getTableNoCache(indexName);
+ List<byte[]> rowKeys = getHBaseRowKeys(pconn, pIndex);
+ assertEquals(50, rowKeys.size());
+ for (byte[] rowKey : rowKeys) {
+ String id = Bytes.toString(rowKey, Bytes.SIZEOF_INT, rowKey.length - Bytes.SIZEOF_INT);
+ int rowIdx = Integer.parseInt(id.replace("row_", ""));
+ assertEquals(findNearestCentroid(vectors[rowIdx], KNOWN_CENTROIDS),
+ extractCentroidId(rowKey));
+ }
+ assertIndexVerifies(tableName, indexName, 50);
+ }
+ }
+
+ /** Verifies transactional constraints reject vector index creation. */
+ @Test
+ public void testVectorIndexRejectedOnTransactionalTable() throws Exception {
+ String txTable = "T_VEC_TX_" + generateUniqueName();
+ String plainTable = "T_VEC_NTX_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ conn.createStatement().execute(
+ "CREATE TABLE " + txTable + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 3))"
+ + " TRANSACTIONAL=true, TRANSACTION_PROVIDER='OMID'");
+ try {
+ conn.createStatement().execute("CREATE VECTOR INDEX IDX_" + generateUniqueName() + " ON "
+ + txTable + " (V) WITH (metric='L2', algorithm='IVF', lists=2, sample_size=10)");
+ fail("CREATE VECTOR INDEX on a transactional table must be rejected");
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.VECTOR_INDEX_ON_TRANSACTIONAL_TABLE.getErrorCode(),
+ e.getErrorCode());
+ }
+
+ conn.createStatement().execute(
+ "CREATE TABLE " + plainTable + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 3))");
+ conn.createStatement().execute("CREATE VECTOR INDEX IDX_" + generateUniqueName() + " ON "
+ + plainTable + " (V) WITH (metric='L2', algorithm='IVF', lists=2, sample_size=10)");
+ try {
+ conn.createStatement().execute(
+ "ALTER TABLE " + plainTable + " SET TRANSACTIONAL=true, TRANSACTION_PROVIDER='OMID'");
+ fail("Making a table with a vector index transactional must be rejected");
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.VECTOR_INDEX_ON_TRANSACTIONAL_TABLE.getErrorCode(),
+ e.getErrorCode());
+ }
+ }
+ }
}
diff --git a/phoenix-core/src/test/java/org/apache/hadoop/hbase/ipc/controller/ServerRpcControllerFactoryTest.java b/phoenix-core/src/test/java/org/apache/hadoop/hbase/ipc/controller/ServerRpcControllerFactoryTest.java
new file mode 100644
index 0000000000..3e2cbcc4f6
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/hadoop/hbase/ipc/controller/ServerRpcControllerFactoryTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.hadoop.hbase.ipc.controller;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.ipc.HBaseRpcController;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.SchemaUtil;
+import org.junit.Test;
+
+public class ServerRpcControllerFactoryTest {
+
+ /**
+ * Verifies that server-to-server calls on SYSTEM.CATALOG and SYSTEM.VECTOR_CENTROID get the
+ * server-side priority, with and without namespace mapping. Index maintenance reads
+ * SYSTEM.VECTOR_CENTROID from RPC handlers on a centroid cache miss. Calls on a user table must
+ * not get this priority.
+ */
+ @Test
+ public void testServerSidePriorityForSystemTables() {
+ Configuration conf = HBaseConfiguration.create();
+ int serverSidePriority = IndexUtil.getServerSidePriority(conf);
+ ServerRpcControllerFactory factory = new ServerRpcControllerFactory(conf);
+ for (byte[] name : new byte[][] { PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES,
+ PhoenixDatabaseMetaData.SYSTEM_VECTOR_CENTROID_NAME_BYTES }) {
+ for (boolean namespaceMapped : new boolean[] { false, true }) {
+ TableName table = SchemaUtil.getPhysicalTableName(name, namespaceMapped);
+ HBaseRpcController controller = factory.newController();
+ controller.setPriority(table);
+ assertEquals(table.getNameAsString(), serverSidePriority, controller.getPriority());
+ }
+ }
+ HBaseRpcController controller = factory.newController();
+ controller.setPriority(TableName.valueOf("T"));
+ assertNotEquals(serverSidePriority, controller.getPriority());
+ }
+}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java
index e4a3db19d6..2a695ee2c5 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.index;
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;
@@ -29,6 +30,8 @@ import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.Time;
import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -36,12 +39,17 @@ import java.util.Properties;
import java.util.Set;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
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.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.cache.VectorCentroidCache;
+import org.apache.phoenix.cache.VectorCentroidCache.CachedCentroids;
+import org.apache.phoenix.coprocessor.generated.ServerCachingProtos;
import org.apache.phoenix.end2end.index.IndexTestUtil;
import org.apache.phoenix.hbase.index.AbstractValueGetter;
import org.apache.phoenix.hbase.index.ValueGetter;
@@ -50,16 +58,22 @@ import org.apache.phoenix.hbase.index.util.GenericKeyValueBuilder;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.optimize.DistanceMetric;
import org.apache.phoenix.query.BaseConnectionlessQueryTest;
import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.DelegateTable;
+import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableKey;
+import org.apache.phoenix.schema.types.PInteger;
+import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.TestUtil;
+import org.apache.phoenix.util.ViewUtil;
import org.junit.Test;
import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
@@ -445,4 +459,235 @@ public class IndexMaintainerTest extends BaseConnectionlessQueryTest {
}
}
}
+
+ /** Makes an IndexMaintainer for a 2D vector index whose centroids are already in the cache. */
+ private IndexMaintainer createVectorIndexMaintainer(PhoenixConnection pconn, String tableName,
+ String indexName, String elementType, long generation) throws Exception {
+ pconn.createStatement().execute("CREATE TABLE " + tableName
+ + " (ID VARCHAR PRIMARY KEY, V VECTOR(" + elementType + ", 2), LABEL VARCHAR)");
+ pconn.createStatement()
+ .execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName
+ + " (V) INCLUDE (LABEL) WITH (algorithm = 'IVF', metric = 'L2', lists = 2,"
+ + " sample_size = 10) ASYNC");
+ PTable dataTable = pconn.getTable(tableName);
+ PTable index = pconn.getTable(indexName);
+ // The client does not send a maintainer for an untrained ASYNC vector index
+ assertNull(index.getVectorCentroidGeneration());
+ assertFalse(IndexMaintainer.sendIndexMaintainer(index));
+ PTable trained = new DelegateTable(index) {
+ @Override
+ public Long getVectorCentroidGeneration() {
+ return generation;
+ }
+ };
+ assertTrue(IndexMaintainer.sendIndexMaintainer(trained));
+ VectorCentroidCache.getInstance(HBaseConfiguration.create()).put(index.getName().getString(),
+ generation, new CachedCentroids(Arrays.asList(new float[] { 0, 0 }, new float[] { 10, 10 }),
+ DistanceMetric.L2));
+ return IndexMaintainer.create(dataTable, trained, pconn);
+ }
+
+ private static Put dataRow(PhoenixConnection pconn, String sql) throws Exception {
+ pconn.createStatement().execute(sql);
+ Iterator<Pair<byte[], List<Mutation>>> iterator = pconn.getMutationState().toMutations();
+ Put put = (Put) iterator.next().getSecond().get(0);
+ pconn.rollback();
+ return put;
+ }
+
+ @Test
+ public void testVectorIndexRowKey() throws Exception {
+ testVectorIndexRowKey("FLOAT");
+ }
+
+ @Test
+ public void testDoubleVectorIndexRowKey() throws Exception {
+ testVectorIndexRowKey("DOUBLE");
+ }
+
+ private void testVectorIndexRowKey(String elementType) throws Exception {
+ String tableName = "T_" + generateUniqueName();
+ String indexName = "I_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ IndexMaintainer im =
+ createVectorIndexMaintainer(pconn, tableName, indexName, elementType, 7L);
+ assertTrue(im.isVectorIndex());
+ assertEquals(DistanceMetric.L2, im.getDistanceMetric());
+ assertEquals(Long.valueOf(7L), im.getCentroidGeneration());
+ // The first row key slot is the centroid ID, a non-null INTEGER
+ assertEquals(PInteger.INSTANCE, im.getIndexRowKeySchema().getField(0).getDataType());
+
+ Put near = dataRow(pconn, "UPSERT INTO " + tableName + " VALUES ('a', ARRAY[9, 8], 'x')");
+ assertTrue(im.shouldPrepareIndexMutations(near));
+ long ts = EnvironmentEdgeManager.currentTimeMillis();
+ ImmutableBytesPtr rowKey = new ImmutableBytesPtr(near.getRow());
+ Put indexPut = im.buildUpdateMutation(GenericKeyValueBuilder.INSTANCE,
+ new IndexUtil.SimpleValueGetter(near), rowKey, ts, null, null, false, null);
+ assertArrayEquals(ByteUtil.concat(PInteger.INSTANCE.toBytes(1), Bytes.toBytes("a")),
+ indexPut.getRow());
+ // The index row also keeps the indexed vector value
+ PColumn vecCol = pconn.getTable(indexName).getColumnForColumnName("0:V");
+ assertEquals(1,
+ indexPut.get(vecCol.getFamilyName().getBytes(), vecCol.getColumnQualifierBytes()).size());
+
+ // The maintainer attributes stay the same after serialization and deserialization
+ IndexMaintainer fromProto = IndexMaintainer.fromProto(IndexMaintainer.toProto(im),
+ pconn.getTable(tableName).getRowKeySchema(), false);
+ assertTrue(fromProto.isVectorIndex());
+ assertEquals(Long.valueOf(7L), fromProto.getCentroidGeneration());
+ assertArrayEquals(indexPut.getRow(),
+ fromProto.buildRowKey(new IndexUtil.SimpleValueGetter(near), rowKey, null, null, ts));
+
+ // A row without a vector has no index row and gets no index mutations
+ Put noVector = dataRow(pconn, "UPSERT INTO " + tableName + " (ID, LABEL) VALUES ('b', 'y')");
+ assertFalse(im.shouldPrepareIndexMutations(noVector));
+ assertNull(im.buildRowKey(new IndexUtil.SimpleValueGetter(noVector),
+ new ImmutableBytesPtr(noVector.getRow()), null, null, ts));
+
+ Put same = dataRow(pconn, "UPSERT INTO " + tableName + " VALUES ('a', ARRAY[9, 8], 'z')");
+ Put moved = dataRow(pconn, "UPSERT INTO " + tableName + " VALUES ('a', ARRAY[1, 0], 'x')");
+ assertTrue(im.isVectorUnchanged(near, same));
+ assertFalse(im.isVectorUnchanged(near, moved));
+ assertFalse(im.isVectorUnchanged(noVector, near));
+ }
+ }
+
+ @Test
+ public void testNonVectorIndexMaintainerHasNoVectorFields() throws Exception {
+ String tableName = "T_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ conn.createStatement()
+ .execute("CREATE TABLE " + tableName + " (ID VARCHAR PRIMARY KEY, C VARCHAR)");
+ conn.createStatement().execute("CREATE INDEX I_" + tableName + " ON " + tableName + " (C)");
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable table = pconn.getTable(tableName);
+ IndexMaintainer im = table.getIndexes().get(0).getIndexMaintainer(table, pconn);
+ ServerCachingProtos.IndexMaintainer proto = IndexMaintainer.toProto(im);
+ assertFalse(proto.hasVectorAlgorithm());
+ assertFalse(IndexMaintainer.fromProto(proto, table.getRowKeySchema(), false).isVectorIndex());
+ }
+ }
+
+ /**
+ * Verifies that the maintainer of an untrained vector index loads no centroids and makes no index
+ * row. A client makes this maintainer for a delete from an immutable table.
+ */
+ @Test
+ public void testUntrainedVectorIndexKeepsNoIndexRows() throws Exception {
+ String tableName = "T_" + generateUniqueName();
+ String indexName = "I_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ pconn.createStatement().execute(
+ "CREATE IMMUTABLE TABLE " + tableName + " (ID VARCHAR PRIMARY KEY, V VECTOR(FLOAT, 2))");
+ pconn.createStatement().execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName
+ + " (V) WITH (algorithm = 'IVF', metric = 'L2', lists = 2, sample_size = 10) ASYNC");
+ PTable index = pconn.getTable(indexName);
+ assertNull(index.getVectorCentroidGeneration());
+ IndexMaintainer im = IndexMaintainer.create(pconn.getTable(tableName), index, pconn);
+ im.loadCentroids(pconn);
+ Put row = dataRow(pconn, "UPSERT INTO " + tableName + " VALUES ('a', ARRAY[9, 8])");
+ ValueGetter getter = new IndexUtil.SimpleValueGetter(row);
+ ImmutableBytesPtr rowKey = new ImmutableBytesPtr(row.getRow());
+ long ts = EnvironmentEdgeManager.currentTimeMillis();
+ assertNull(im.buildRowKey(getter, rowKey, null, null, ts));
+ assertNull(im.buildUpdateMutation(GenericKeyValueBuilder.INSTANCE, getter, rowKey, ts, null,
+ null, false));
+ }
+ }
+
+ /**
+ * Verifies that a client keeps the centroids that it loaded before a write. The client has no
+ * server connection, so it cannot reload centroids that the cache evicts.
+ */
+ @Test
+ public void testClientHoldsLoadedCentroids() throws Exception {
+ String tableName = "T_" + generateUniqueName();
+ String indexName = "I_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ IndexMaintainer im = createVectorIndexMaintainer(pconn, tableName, indexName, "FLOAT", 7L);
+ im.loadCentroids(pconn);
+ VectorCentroidCache.getInstance(HBaseConfiguration.create()).invalidate(indexName);
+ Put near = dataRow(pconn, "UPSERT INTO " + tableName + " VALUES ('a', ARRAY[9, 8], 'x')");
+ Put indexPut = im.buildUpdateMutation(GenericKeyValueBuilder.INSTANCE,
+ new IndexUtil.SimpleValueGetter(near), new ImmutableBytesPtr(near.getRow()),
+ EnvironmentEdgeManager.currentTimeMillis(), null, null, false, null);
+ assertArrayEquals(ByteUtil.concat(PInteger.INSTANCE.toBytes(1), Bytes.toBytes("a")),
+ indexPut.getRow());
+ }
+ }
+
+ /**
+ * Verifies that a maintainer keeps the centroids that it resolved for its first row. A model that
+ * the cache evicts, or a model larger than the cache, must not reload for each later row.
+ */
+ @Test
+ public void testMaintainerHoldsResolvedCentroids() throws Exception {
+ String tableName = "T_" + generateUniqueName();
+ String indexName = "I_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ IndexMaintainer im = createVectorIndexMaintainer(pconn, tableName, indexName, "FLOAT", 7L);
+ Put near = dataRow(pconn, "UPSERT INTO " + tableName + " VALUES ('a', ARRAY[9, 8], 'x')");
+ Put far = dataRow(pconn, "UPSERT INTO " + tableName + " VALUES ('b', ARRAY[1, 0], 'x')");
+ long ts = EnvironmentEdgeManager.currentTimeMillis();
+ assertArrayEquals(ByteUtil.concat(PInteger.INSTANCE.toBytes(1), Bytes.toBytes("a")),
+ im.buildRowKey(new IndexUtil.SimpleValueGetter(near), new ImmutableBytesPtr(near.getRow()),
+ null, null, ts));
+ // This process has no server connection, so a reload fails
+ VectorCentroidCache.getInstance(HBaseConfiguration.create()).invalidate(indexName);
+ assertArrayEquals(ByteUtil.concat(PInteger.INSTANCE.toBytes(0), Bytes.toBytes("b")),
+ im.buildRowKey(new IndexUtil.SimpleValueGetter(far), new ImmutableBytesPtr(far.getRow()),
+ null, null, ts));
+ }
+ }
+
+ /**
+ * Verifies that a view with a WHERE clause inherits a vector index with a name that has a view
+ * prefix. The inherited index must assign rows with the centroids of the parent index.
+ */
+ @Test
+ public void testViewInheritedVectorIndexUsesIndexCentroids() throws Exception {
+ String tableName = "T_" + generateUniqueName();
+ String indexName = "I_" + generateUniqueName();
+ String viewName = "V_" + generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ pconn.createStatement().execute(
+ "CREATE TABLE " + tableName + " (ID VARCHAR PRIMARY KEY, K VARCHAR, V VECTOR(FLOAT, 2))");
+ // A view inherits an index only if the index covers the WHERE columns of the view
+ pconn.createStatement()
+ .execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName
+ + " (V) INCLUDE (K) WITH (algorithm = 'IVF', metric = 'L2', lists = 2, sample_size = 10)"
+ + " ASYNC");
+ pconn.createStatement()
+ .execute("CREATE VIEW " + viewName + " AS SELECT * FROM " + tableName + " WHERE K = 'k'");
+ // Inherit the index in the same way as the view resolution from SYSTEM.CATALOG
+ PTable view = pconn.getTable(viewName);
+ List<PTable> inheritedIndexes = new ArrayList<>();
+ ViewUtil.addIndexesFromParent(pconn, view, pconn.getTable(tableName), inheritedIndexes);
+ PTable inherited = inheritedIndexes.get(0);
+ assertEquals(viewName + QueryConstants.CHILD_VIEW_INDEX_NAME_SEPARATOR + indexName,
+ inherited.getName().getString());
+ PTable trained = new DelegateTable(inherited) {
+ @Override
+ public Long getVectorCentroidGeneration() {
+ return 5L;
+ }
+ };
+ VectorCentroidCache.getInstance(HBaseConfiguration.create()).put(indexName, 5L,
+ new CachedCentroids(Arrays.asList(new float[] { 0, 0 }, new float[] { 10, 10 }),
+ DistanceMetric.L2));
+ IndexMaintainer im = IndexMaintainer.create(view, trained, pconn);
+ im.loadCentroids(pconn);
+ Put near = dataRow(pconn, "UPSERT INTO " + viewName + " (ID, V) VALUES ('a', ARRAY[9, 8])");
+ Put indexPut = im.buildUpdateMutation(GenericKeyValueBuilder.INSTANCE,
+ new IndexUtil.SimpleValueGetter(near), new ImmutableBytesPtr(near.getRow()),
+ EnvironmentEdgeManager.currentTimeMillis(), null, null, false, null);
+ assertArrayEquals(ByteUtil.concat(PInteger.INSTANCE.toBytes(1), Bytes.toBytes("a")),
+ indexPut.getRow());
+ }
+ }
}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/index/vector/CentroidManagerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/index/vector/CentroidManagerTest.java
index 4ef8c4508a..3a870a925b 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/index/vector/CentroidManagerTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/index/vector/CentroidManagerTest.java
@@ -62,4 +62,15 @@ public class CentroidManagerTest {
setClock(2_000L);
assertEquals(2_000L, CentroidManager.nextGeneration(1_000L));
}
+
+ /**
+ * Verifies that the name of an index that a view inherits resolves to the name of the parent
+ * index. The inherited name can have prefixes for the view and its ancestor views.
+ */
+ @Test
+ public void testInheritedIndexUsesIndexCentroids() {
+ assertEquals("S.IDX", CentroidManager.getCentroidIndexName("S.IDX"));
+ assertEquals("IDX", CentroidManager.getCentroidIndexName("V#IDX"));
+ assertEquals("S.IDX", CentroidManager.getCentroidIndexName("S.VS.V2#VS.V1#S.IDX"));
+ }
}