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 <[email protected]> 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) <[email protected]> --- .../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")); + } }
