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 509ce3014eb0bf5730cccb1c0594c8be05e8e72e Author: Andrew Purtell <[email protected]> AuthorDate: Tue Oct 6 00:01:09 2026 -0700 PHOENIX-7998 Phase 10: BSON vector extraction Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> --- .../apache/phoenix/compile/ProjectionCompiler.java | 80 ++- .../BaseScannerRegionObserverConstants.java | 1 + .../apache/phoenix/expression/ExpressionType.java | 3 +- .../function/BsonVectorValueFunction.java | 232 ++++++++ .../org/apache/phoenix/index/IndexMaintainer.java | 209 ++++++-- .../org/apache/phoenix/iterate/ExplainTable.java | 3 +- .../apache/phoenix/optimize/OptimizerReasons.java | 2 + .../apache/phoenix/optimize/QueryOptimizer.java | 19 +- .../phoenix/parse/BsonVectorValueParseNode.java | 54 ++ .../parse/IndexExpressionParseNodeRewriter.java | 58 +- .../src/main/protobuf/ServerCachingService.proto | 1 + .../phoenix/hbase/index/IndexRegionObserver.java | 25 +- .../iterate/NonAggregateRegionScannerFactory.java | 9 +- .../phoenix/iterate/RegionScannerFactory.java | 19 +- .../phoenix/end2end/BsonServerProjectionIT.java | 171 ++++++ .../apache/phoenix/end2end/VectorBsonIndexIT.java | 589 +++++++++++++++++++++ .../apache/phoenix/end2end/VectorIvfQueryIT.java | 6 +- .../function/BsonVectorValueFunctionTest.java | 332 ++++++++++++ .../apache/phoenix/index/IndexMaintainerTest.java | 209 +++++++- .../phoenix/iterate/RegionScannerFactoryTest.java | 80 +++ .../phoenix/parse/VectorColumnParseTest.java | 15 + 21 files changed, 2032 insertions(+), 85 deletions(-) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java index 90844b5aac..ddc8f2c3a3 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java @@ -27,9 +27,11 @@ import java.sql.SQLException; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.NavigableSet; +import java.util.Set; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.util.Bytes; @@ -47,6 +49,7 @@ import org.apache.phoenix.expression.ProjectedColumnExpression; import org.apache.phoenix.expression.SingleCellColumnExpression; import org.apache.phoenix.expression.function.ArrayIndexFunction; import org.apache.phoenix.expression.function.BsonValueFunction; +import org.apache.phoenix.expression.function.BsonVectorValueFunction; import org.apache.phoenix.expression.function.JsonQueryFunction; import org.apache.phoenix.expression.function.JsonValueFunction; import org.apache.phoenix.expression.visitor.ExpressionVisitor; @@ -58,10 +61,13 @@ import org.apache.phoenix.parse.BindParseNode; import org.apache.phoenix.parse.ColumnParseNode; import org.apache.phoenix.parse.FamilyWildcardParseNode; import org.apache.phoenix.parse.FunctionParseNode; +import org.apache.phoenix.parse.LiteralParseNode; +import org.apache.phoenix.parse.OrderByNode; import org.apache.phoenix.parse.ParseNode; import org.apache.phoenix.parse.PhoenixRowTimestampParseNode; import org.apache.phoenix.parse.SelectStatement; import org.apache.phoenix.parse.SequenceValueParseNode; +import org.apache.phoenix.parse.StatelessTraverseAllParseNodeVisitor; import org.apache.phoenix.parse.TableName; import org.apache.phoenix.parse.TableWildcardParseNode; import org.apache.phoenix.parse.WildcardParseNode; @@ -540,10 +546,13 @@ public class ProjectionCompiler { index++; } + // Do not evaluate a function on the server if ORDER BY references the column of the function. + // The ORDER BY evaluation must read the original column cell in the row. + Set<String> orderByColumnNames = getOrderByColumnNames(statement); for (int i = serverParsedProjectedColumnRefs.size() - 1; i >= 0; i--) { - Expression expression = serverParsedProjectedColumnRefs.get(i); + ProjectedColumnExpression expression = serverParsedProjectedColumnRefs.get(i); Integer count = serverParsedExpressionCounts.get(expression); - if (count != 0) { + if (count != 0 || orderByColumnNames.contains(expression.getColumn().getName().getString())) { serverParsedKVRefs.remove(i); serverParsedKVFuncs.remove(i); serverParsedOldFuncs.remove(i); @@ -551,17 +560,21 @@ public class ProjectionCompiler { } if (serverParsedKVFuncs.size() > 0 && serverParsedKVRefs.size() > 0) { + // This order must match the order in which the server evaluates the server parsed functions. + // See NonAggregateRegionScannerFactory#getServerParsedExpressions. String[] scanAttributes = new String[] { BaseScannerRegionObserverConstants.SPECIFIC_ARRAY_INDEX, BaseScannerRegionObserverConstants.JSON_VALUE_FUNCTION, + BaseScannerRegionObserverConstants.BSON_VALUE_FUNCTION, BaseScannerRegionObserverConstants.JSON_QUERY_FUNCTION, - BaseScannerRegionObserverConstants.BSON_VALUE_FUNCTION }; + BaseScannerRegionObserverConstants.BSON_VECTOR_VALUE_FUNCTION }; Map<String, Class> attributeToFunctionMap = new HashMap<String, Class>() { { put(scanAttributes[0], ArrayIndexFunction.class); put(scanAttributes[1], JsonValueFunction.class); - put(scanAttributes[2], JsonQueryFunction.class); - put(scanAttributes[3], BsonValueFunction.class); + put(scanAttributes[2], BsonValueFunction.class); + put(scanAttributes[3], JsonQueryFunction.class); + put(scanAttributes[4], BsonVectorValueFunction.class); } }; // This map is to keep track of the positions that get swapped with rearranging @@ -603,17 +616,17 @@ public class ProjectionCompiler { // Stash the per-type expression buckets on the context so EXPLAIN can render the per-type // SERVER ARRAY|JSON|BSON PROJECTION clauses (and their per-expression detail lines). context.setServerParsedProjections(serverAttributeToFuncExpressionMap); - KeyValueSchemaBuilder builder = new KeyValueSchemaBuilder(0); - for (Expression expression : serverParsedKVRefs) { - builder.addField(expression); + // Build the schema in the field order that the server uses to serialize the projected values + Expression[] shuffledFuncs = new Expression[serverParsedKVFuncs.size()]; + for (int i = 0; i < serverParsedKVFuncs.size(); i++) { + shuffledFuncs[initialToShuffledPositionMap.get(i)] = serverParsedKVFuncs.get(i); } - KeyValueSchema kvSchema = builder.build(); - ValueBitSet arrayIndexesBitSet = ValueBitSet.newInstance(kvSchema); - builder = new KeyValueSchemaBuilder(0); - for (Expression expression : serverParsedKVFuncs) { + KeyValueSchemaBuilder builder = new KeyValueSchemaBuilder(0); + for (Expression expression : shuffledFuncs) { builder.addField(expression); } KeyValueSchema arrayIndexesSchema = builder.build(); + ValueBitSet arrayIndexesBitSet = ValueBitSet.newInstance(arrayIndexesSchema); Map<Expression, Expression> replacementMap = new HashMap<>(); for (int i = 0; i < serverParsedOldFuncs.size(); i++) { @@ -925,12 +938,53 @@ public class ProjectionCompiler { } } + /** + * Returns the names of the columns that the ORDER BY clause references. A column alias or an + * ordinal position resolves to its select expression. + */ + private static Set<String> getOrderByColumnNames(SelectStatement statement) throws SQLException { + Set<String> names = new HashSet<>(); + if (statement.getOrderBy().isEmpty()) { + return names; + } + Map<String, ParseNode> aliases = new HashMap<>(); + for (AliasedNode aliasedNode : statement.getSelect()) { + if (aliasedNode.getAlias() != null) { + aliases.put(aliasedNode.getAlias(), aliasedNode.getNode()); + } + } + StatelessTraverseAllParseNodeVisitor visitor = new StatelessTraverseAllParseNodeVisitor() { + @Override + public Void visit(ColumnParseNode node) { + names.add(node.getName()); + return null; + } + }; + for (OrderByNode orderByNode : statement.getOrderBy()) { + ParseNode node = orderByNode.getNode(); + if (node instanceof ColumnParseNode && ((ColumnParseNode) node).getTableName() == null) { + ParseNode aliased = aliases.get(((ColumnParseNode) node).getName()); + node = aliased != null ? aliased : node; + } else if ( + node instanceof LiteralParseNode && ((LiteralParseNode) node).getValue() instanceof Integer + ) { + int position = (Integer) ((LiteralParseNode) node).getValue(); + if (position >= 1 && position <= statement.getSelect().size()) { + node = statement.getSelect().get(position - 1).getNode(); + } + } + node.accept(visitor); + } + return names; + } + private static boolean isJsonFunction(FunctionParseNode node) { return JsonValueFunction.NAME.equals(node.getName()) || JsonQueryFunction.NAME.equals(node.getName()); } private static boolean isBsonFunction(FunctionParseNode node) { - return BsonValueFunction.NAME.equals(node.getName()); + return BsonValueFunction.NAME.equals(node.getName()) + || BsonVectorValueFunction.NAME.equals(node.getName()); } } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/BaseScannerRegionObserverConstants.java b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/BaseScannerRegionObserverConstants.java index 3badbbdebf..cbdfb7e535 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/BaseScannerRegionObserverConstants.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/BaseScannerRegionObserverConstants.java @@ -106,6 +106,7 @@ public class BaseScannerRegionObserverConstants { public static final String JSON_VALUE_FUNCTION = "_JsonValueFunction"; public static final String JSON_QUERY_FUNCTION = "_JsonQueryFunction"; public static final String BSON_VALUE_FUNCTION = "_BsonValueFunction"; + public static final String BSON_VECTOR_VALUE_FUNCTION = "_BsonVectorValueFunction"; /* * Attribute to denote that the index maintainer has been serialized using its proto-buf diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/expression/ExpressionType.java b/phoenix-core-client/src/main/java/org/apache/phoenix/expression/ExpressionType.java index c857a6f2e6..f519b12a66 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/expression/ExpressionType.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/expression/ExpressionType.java @@ -215,7 +215,8 @@ public enum ExpressionType { L2DistanceFunction(L2DistanceFunction.class), L2DistanceSquaredFunction(L2DistanceSquaredFunction.class), CosineDistanceFunction(CosineDistanceFunction.class), - InnerProductDistanceFunction(InnerProductDistanceFunction.class); + InnerProductDistanceFunction(InnerProductDistanceFunction.class), + BsonVectorValueFunction(BsonVectorValueFunction.class); ExpressionType(Class<? extends Expression> clazz) { this.clazz = clazz; diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/expression/function/BsonVectorValueFunction.java b/phoenix-core-client/src/main/java/org/apache/phoenix/expression/function/BsonVectorValueFunction.java new file mode 100644 index 0000000000..3d2fb0faca --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/expression/function/BsonVectorValueFunction.java @@ -0,0 +1,232 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.expression.function; + +import java.io.DataInput; +import java.io.IOException; +import java.util.List; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.phoenix.exception.SQLExceptionCode; +import org.apache.phoenix.exception.SQLExceptionInfo; +import org.apache.phoenix.expression.Expression; +import org.apache.phoenix.expression.LiteralExpression; +import org.apache.phoenix.expression.util.bson.CommonComparisonExpressionUtils; +import org.apache.phoenix.parse.BsonVectorValueParseNode; +import org.apache.phoenix.parse.FunctionParseNode; +import org.apache.phoenix.schema.IllegalDataException; +import org.apache.phoenix.schema.tuple.Tuple; +import org.apache.phoenix.schema.types.PBson; +import org.apache.phoenix.schema.types.PDataType; +import org.apache.phoenix.schema.types.PInteger; +import org.apache.phoenix.schema.types.PJson; +import org.apache.phoenix.schema.types.PVarbinary; +import org.apache.phoenix.schema.types.PVarchar; +import org.apache.phoenix.schema.types.PVectorFloat; +import org.apache.phoenix.util.ByteUtil; +import org.bson.BinaryVector; +import org.bson.BsonBinary; +import org.bson.BsonBinarySubType; +import org.bson.BsonDocument; +import org.bson.BsonValue; + +import org.apache.phoenix.thirdparty.com.google.common.base.Preconditions; + +/** + * Function extracting a typed float vector from a BSON document field. + * <p> + * Arguments: + * <ol> + * <li>BSON document expression (PBson, PJson, or PVarbinary)</li> + * <li>Target field path literal</li> + * <li>Vector dimension literal</li> + * </ol> + * </p> + * Transcodes BSON Binary subtype 9 (VECTOR) FLOAT32 payloads from little endian format into the big + * endian packed representation of {@link PVectorFloat}. Missing paths or documents evaluate to SQL + * NULL. Structural or dimension mismatches raise {@link IllegalDataException}. + */ [email protected](name = BsonVectorValueFunction.NAME, + nodeClass = BsonVectorValueParseNode.class, + args = { + @FunctionParseNode.Argument(allowedTypes = { PJson.class, PBson.class, PVarbinary.class }), + @FunctionParseNode.Argument(allowedTypes = { PVarchar.class }, isConstant = true), + @FunctionParseNode.Argument(allowedTypes = { PInteger.class }, isConstant = true) }) +public class BsonVectorValueFunction extends ScalarFunction { + + public static final String NAME = "BSON_VECTOR_VALUE"; + + /** + * The size in bytes of the BSON binary vector header: one data type byte and one padding byte. + */ + public static final int VECTOR_HEADER_SIZE = 2; + + private static final byte FLOAT32_DTYPE = BinaryVector.DataType.FLOAT32.getValue(); + + // The constant path and dimension arguments, resolved at construction and after deserialization + private String path; + private int dimension; + + public BsonVectorValueFunction() { + // For deserialization. readFields resolves the arguments. + } + + public BsonVectorValueFunction(List<Expression> children) { + super(children); + Preconditions.checkNotNull(children, "children cannot be null"); + Preconditions.checkArgument(children.size() >= 3, + "BSON_VECTOR_VALUE requires 3 arguments: document, path, and dimension"); + Preconditions.checkNotNull(children.get(0), "document expression cannot be null"); + Preconditions.checkNotNull(children.get(1), "path expression cannot be null"); + Preconditions.checkNotNull(children.get(2), "dimension expression cannot be null"); + init(); + } + + private void init() { + this.path = evaluateConstant(children.get(1), PVarchar.INSTANCE, String.class, "path"); + Number dim = evaluateConstant(children.get(2), PInteger.INSTANCE, Number.class, "dimension"); + this.dimension = dim.intValue(); + if (dimension <= 0) { + throw new IllegalArgumentException( + "Dimension must be a positive integer, but got: " + dimension); + } + if (dimension > PVectorFloat.MAX_VECTOR_DIMENSION) { + throw new IllegalArgumentException("Dimension exceeds maximum allowed vector dimension: " + + dimension + " > " + PVectorFloat.MAX_VECTOR_DIMENSION); + } + } + + private static <T> T evaluateConstant(Expression expr, PDataType<?> type, Class<T> clazz, + String argName) { + Object val = null; + if (expr instanceof LiteralExpression) { + val = ((LiteralExpression) expr).getValue(); + } else { + ImmutableBytesWritable ptr = new ImmutableBytesWritable(); + if (expr.evaluate(null, ptr) && ptr.getLength() > 0) { + val = type.toObject(ptr, expr.getDataType(), expr.getSortOrder()); + } + } + if (!clazz.isInstance(val)) { + throw new IllegalArgumentException( + "BSON_VECTOR_VALUE " + argName + " argument must be a non-null constant"); + } + return clazz.cast(val); + } + + @Override + public void readFields(DataInput input) throws IOException { + super.readFields(input); + init(); + } + + public int getDimension() { + return dimension; + } + + public String getPath() { + return path; + } + + @Override + public String getName() { + return NAME; + } + + @Override + public PDataType<?> getDataType() { + return PVectorFloat.INSTANCE; + } + + @Override + public Integer getMaxLength() { + return dimension; + } + + @Override + public boolean evaluate(Tuple tuple, ImmutableBytesWritable ptr) { + if (!getChildren().get(0).evaluate(tuple, ptr)) { + return false; + } + if (ptr.getLength() == 0) { + return true; + } + Object object = PBson.INSTANCE.toObject(ptr, getChildren().get(0).getSortOrder()); + if (!(object instanceof BsonDocument)) { + throw malformed("expected a BSON document, but found " + + (object == null ? "null" : object.getClass().getSimpleName())); + } + BsonValue bsonValue = + CommonComparisonExpressionUtils.getFieldFromDocument(path, (BsonDocument) object); + if (bsonValue == null || bsonValue.isNull()) { + ptr.set(ByteUtil.EMPTY_BYTE_ARRAY); + return true; + } + if (!(bsonValue instanceof BsonBinary)) { + throw malformed("expected a BSON Binary value, but found " + bsonValue.getBsonType()); + } + BsonBinary binary = (BsonBinary) bsonValue; + if (binary.getType() != BsonBinarySubType.VECTOR.getValue()) { + throw malformed("expected BSON Binary subtype " + BsonBinarySubType.VECTOR.getValue() + + " (VECTOR), but found subtype " + binary.getType()); + } + byte[] data = binary.getData(); + if (data.length < VECTOR_HEADER_SIZE) { + throw malformed("truncated vector of " + data.length + " bytes"); + } + if (data[0] != FLOAT32_DTYPE) { + throw malformed("expected FLOAT32 (0x" + Integer.toHexString(FLOAT32_DTYPE & 0xFF) + + ") elements, but found 0x" + Integer.toHexString(data[0] & 0xFF)); + } + if (data[1] != 0) { + throw malformed("expected zero padding for FLOAT32, but found " + (data[1] & 0xFF)); + } + // A NaN or infinite element makes the vector malformed. No distance can rank such a vector, and + // it can corrupt centroid training. + for (int i = 0; i < (data.length - VECTOR_HEADER_SIZE) / Bytes.SIZEOF_FLOAT; i++) { + float element = Float.intBitsToFloat( + Integer.reverseBytes(Bytes.toInt(data, VECTOR_HEADER_SIZE + i * Bytes.SIZEOF_FLOAT))); + if (!Float.isFinite(element)) { + throw malformed("element " + i + " is " + element); + } + } + int payloadLength = data.length - VECTOR_HEADER_SIZE; + if (payloadLength != dimension * Bytes.SIZEOF_FLOAT) { + throw new IllegalDataException( + new SQLExceptionInfo.Builder(SQLExceptionCode.VECTOR_DIMENSION_MISMATCH) + .setMessage("BSON vector dimension mismatch at path '" + path + "': expected " + dimension + + ", found " + payloadLength / (double) Bytes.SIZEOF_FLOAT) + .build().buildException()); + } + // Convert little endian BSON float32 payload to big endian PVectorFloat encoding + byte[] out = new byte[payloadLength]; + for (int i = 0; i < payloadLength; i += Bytes.SIZEOF_FLOAT) { + int src = VECTOR_HEADER_SIZE + i; + out[i] = data[src + 3]; + out[i + 1] = data[src + 2]; + out[i + 2] = data[src + 1]; + out[i + 3] = data[src]; + } + ptr.set(out); + return true; + } + + private IllegalDataException malformed(String detail) { + return new IllegalDataException("Malformed BSON vector at path '" + path + "': " + detail); + } +} 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 ee3cf88ffc..b1b5e238c5 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 @@ -93,6 +93,7 @@ import org.apache.phoenix.query.QueryConstants; import org.apache.phoenix.schema.AmbiguousColumnException; import org.apache.phoenix.schema.ColumnFamilyNotFoundException; import org.apache.phoenix.schema.ColumnNotFoundException; +import org.apache.phoenix.schema.IllegalDataException; import org.apache.phoenix.schema.PColumn; import org.apache.phoenix.schema.PColumnFamily; import org.apache.phoenix.schema.PDatum; @@ -494,6 +495,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { // eviction cannot cause a reload for each row, or a failure on a client with no server connection private CachedCentroids activeCentroids; private Long centroidGeneration; + // Index column reference for expression derived vector values, or null if directly mapped. + private ColumnReference functionalVectorColumn; protected IndexMaintainer(RowKeySchema dataRowKeySchema, boolean isDataTableSalted) { this.dataRowKeySchema = dataRowKeySchema; @@ -524,6 +527,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { this.vectorAlgorithm = index.getVectorIndexAlgorithm(); this.distanceMetric = DistanceMetric.fromString(index.getVectorDistanceMetric()); this.centroidGeneration = index.getVectorCentroidGeneration(); + PColumn vectorColumn = VectorIndexTrainer.getIndexedVectorColumn(index); + if (IndexUtil.getDataColumnOrNull(dataTable, vectorColumn.getName().getString()) == null) { + this.functionalVectorColumn = new ColumnReference(vectorColumn.getFamilyName().getBytes(), + vectorColumn.getColumnQualifierBytes()); + } } // null check for b/w compatibility @@ -796,8 +804,33 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { return buildRowKey(valueGetter, rowKeyPtr, regionStartKey, regionEndKey, ts, null); } + /** + * Builds the index row key for a data row. For a vector index, this method evaluates the vector + * leniently. It returns null if the vector is null or malformed, or if the index has no trained + * centroids. + */ public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr, byte[] regionStartKey, byte[] regionEndKey, long ts, byte[] encodedRegionName) { + return buildRowKey(valueGetter, rowKeyPtr, regionStartKey, regionEndKey, ts, encodedRegionName, + null); + } + + /** + * Builds the index row key for a data row from the indexed vector that + * {@link #getIndexedVector(Put, boolean)} already evaluated. Returns null for a vector index if + * the vector is null or if the index has no trained centroids. + */ + public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr, long ts, + byte[] encodedRegionName, ImmutableBytesWritable indexedVector) { + if (isVectorIndex() && indexedVector == null) { + return null; + } + return buildRowKey(valueGetter, rowKeyPtr, null, null, ts, encodedRegionName, indexedVector); + } + + private byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr, + byte[] regionStartKey, byte[] regionEndKey, long ts, byte[] encodedRegionName, + ImmutableBytesWritable indexedVector) { if (isCDCIndex && encodedRegionName == null) { throw new IllegalArgumentException("Encoded region name is required for a CDC index"); } @@ -883,16 +916,13 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { // 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 - ) { + ImmutableBytesWritable vector = + indexedVector != null ? indexedVector : getIndexedVector(valueGetter, ts, false); + if (vector == null) { return null; } - ptr.set(PInteger.INSTANCE.toBytes( - getCentroids().assign(ptr.get(), ptr.getOffset(), ptr.getLength(), dataColumnType))); + ptr.set(PInteger.INSTANCE.toBytes(getCentroids().assign(vector.get(), + vector.getOffset(), vector.getLength(), dataColumnType))); dataColumnType = PInteger.INSTANCE; dataSortOrder = SortOrder.ASC; isNullable = false; @@ -1174,7 +1204,25 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { */ public boolean shouldPrepareIndexMutations(Put dataRowState) { - if (isVectorIndex() && getIndexedVector(dataRowState) == null) { + return shouldPrepareIndexMutations(dataRowState, getIndexedVector(dataRowState, true)); + } + + /** + * Checks whether the current row state corresponds to an existing index entry for deletion. + */ + public boolean hasIndexRow(Put currentDataRowState) { + return shouldPrepareIndexMutations(currentDataRowState, + getIndexedVector(currentDataRowState, false)); + } + + /** + * Determines if the index row for a data row must be prepared. The caller gives the indexed + * vector that {@link #getIndexedVector(Put, boolean)} already evaluated. A vector index prepares + * no index row if the vector is null. Other indexes ignore the vector. + */ + public boolean shouldPrepareIndexMutations(Put dataRowState, + ImmutableBytesWritable indexedVector) { + if (isVectorIndex() && indexedVector == null) { // A row with a null vector has no index row and gets no index mutations return false; } @@ -1385,15 +1433,34 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr, long ts, byte[] regionStartKey, byte[] regionEndKey, boolean verified, byte[] encodedRegionName) throws IOException { + // Evaluate the vector strictly, so that a malformed vector makes the row write fail + return buildUpdateMutation(kvBuilder, valueGetter, dataRowKeyPtr, ts, regionStartKey, + regionEndKey, verified, encodedRegionName, + isVectorIndex() ? getIndexedVector(valueGetter, ts, true) : null); + } + + /** + * Builds the index row update for a data row from the indexed vector that + * {@link #getIndexedVector(Put, boolean)} already evaluated. Returns null for a vector index if + * the vector is null or if the index has no trained centroids. + */ + public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, + ImmutableBytesWritable dataRowKeyPtr, long ts, byte[] regionStartKey, byte[] regionEndKey, + boolean verified, byte[] encodedRegionName, ImmutableBytesWritable indexedVector) + throws IOException { + if (isVectorIndex() && indexedVector == null) { + return null; + } byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey, - ts, encodedRegionName); + ts, encodedRegionName, indexedVector); if (indexRowKey == null) { return null; } return buildUpdateMutation(kvBuilder, valueGetter, dataRowKeyPtr, ts, regionStartKey, regionEndKey, indexRowKey, this.getEmptyKeyValueFamily(), coveredColumnsMap, indexEmptyKeyValueRef, indexWALDisabled, dataImmutableStorageScheme, immutableStorageScheme, - encodingScheme, dataEncodingScheme, verified); + encodingScheme, dataEncodingScheme, verified, functionalVectorColumn, + functionalVectorColumn == null ? null : indexedVector); } public static Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, @@ -1403,6 +1470,20 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { boolean destWALDisabled, ImmutableStorageScheme srcImmutableStorageScheme, ImmutableStorageScheme destImmutableStorageScheme, QualifierEncodingScheme destEncodingScheme, QualifierEncodingScheme srcEncodingScheme, boolean verified) throws IOException { + return buildUpdateMutation(kvBuilder, valueGetter, dataRowKeyPtr, ts, regionStartKey, + regionEndKey, destRowKey, emptyKeyValueCFPtr, coveredColumnsMap, destEmptyKeyValueRef, + destWALDisabled, srcImmutableStorageScheme, destImmutableStorageScheme, destEncodingScheme, + srcEncodingScheme, verified, null, null); + } + + public static Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, + ImmutableBytesWritable dataRowKeyPtr, long ts, byte[] regionStartKey, byte[] regionEndKey, + byte[] destRowKey, ImmutableBytesPtr emptyKeyValueCFPtr, + Map<ColumnReference, ColumnReference> coveredColumnsMap, ColumnReference destEmptyKeyValueRef, + boolean destWALDisabled, ImmutableStorageScheme srcImmutableStorageScheme, + ImmutableStorageScheme destImmutableStorageScheme, QualifierEncodingScheme destEncodingScheme, + QualifierEncodingScheme srcEncodingScheme, boolean verified, ColumnReference computedColumn, + ImmutableBytesWritable computedValue) throws IOException { Set<ColumnReference> coveredColumns = coveredColumnsMap.keySet(); Put put = null; // New row being inserted: add the empty key value @@ -1442,6 +1523,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { } familyToColListMap.get(cf).add(Pair.newPair(indexColRef, ref)); } + if (computedValue != null) { + // The computed vector column has no source column in the data table + familyToColListMap.computeIfAbsent(new ImmutableBytesPtr(computedColumn.getFamily()), + k -> Lists.newArrayList()).add(Pair.newPair(computedColumn, (ColumnReference) null)); + } // iterate over each column family and create a byte[] containing all the columns for (Entry<ImmutableBytesPtr, List<Pair<ColumnReference, ColumnReference>>> entry : familyToColListMap.entrySet()) { @@ -1460,7 +1546,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { ColumnReference indexColRef = colRefPair.getFirst(); ColumnReference dataColRef = colRefPair.getSecond(); byte[] value = null; - if (srcImmutableStorageScheme == ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS) { + if (dataColRef == null) { + value = computedValue.copyBytes(); + } else if ( + srcImmutableStorageScheme == ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS + ) { Expression expression = new SingleCellColumnExpression(new PDatum() { @Override public boolean isNullable() { @@ -1610,6 +1700,14 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { } } } + if (computedValue != null) { + if (put == null) { + put = new Put(destRowKey); + put.setDurability(!destWALDisabled ? Durability.USE_DEFAULT : Durability.SKIP_WAL); + } + put.add(kvBuilder.buildPut(rowKey, computedColumn.getFamilyWritable(), + computedColumn.getQualifierWritable(), ts, computedValue)); + } } return put; } @@ -1634,6 +1732,13 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { } int colsSet = indexUpdate.getFamilyCellMap().values().stream().mapToInt(elem -> elem.size()).sum(); + if ( + functionalVectorColumn != null && indexUpdate.has(functionalVectorColumn.getFamily(), + functionalVectorColumn.getQualifier()) + ) { + // The computed vector column is not a covered column, so do not count it + colsSet--; + } if (coveredColumnsMap.size() + 1 == colsSet) { // add 1 for the empty column // Index row update is always a full update except when some columns are explicitly // set to null. Do a quick size check to determine if some covered columns are being @@ -2148,7 +2253,13 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { if (proto.hasVectorAlgorithm()) { maintainer.vectorAlgorithm = proto.getVectorAlgorithm(); maintainer.distanceMetric = DistanceMetric.fromString(proto.getDistanceMetric()); - maintainer.centroidGeneration = proto.getCentroidGeneration(); + maintainer.centroidGeneration = + proto.hasCentroidGeneration() ? proto.getCentroidGeneration() : null; + if (proto.hasFunctionalVectorColumn()) { + maintainer.functionalVectorColumn = + new ColumnReference(proto.getFunctionalVectorColumn().getFamily().toByteArray(), + proto.getFunctionalVectorColumn().getQualifier().toByteArray()); + } } maintainer.initCachedState(); return maintainer; @@ -2312,7 +2423,14 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { if (maintainer.isVectorIndex()) { builder.setVectorAlgorithm(maintainer.vectorAlgorithm); builder.setDistanceMetric(maintainer.distanceMetric.name()); - builder.setCentroidGeneration(maintainer.centroidGeneration); + if (maintainer.centroidGeneration != null) { + builder.setCentroidGeneration(maintainer.centroidGeneration); + } + if (maintainer.functionalVectorColumn != null) { + builder.setFunctionalVectorColumn(ServerCachingProtos.ColumnReference.newBuilder() + .setFamily(ByteStringer.wrap(maintainer.functionalVectorColumn.getFamily())) + .setQualifier(ByteStringer.wrap(maintainer.functionalVectorColumn.getQualifier()))); + } } return builder.build(); } @@ -2713,6 +2831,14 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { return centroidGeneration; } + /** + * Returns the index column that holds the computed vector of a functional vector index. Returns + * null if the index is not a vector index or indexes a plain vector column. + */ + public ColumnReference getFunctionalVectorColumn() { + return functionalVectorColumn; + } + /** * 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 @@ -2743,33 +2869,54 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { return activeCentroids; } - /** Extracts the indexed vector value from the row state, or null if absent. */ - private ImmutableBytesWritable getIndexedVector(Put dataRowState) { - if (dataRowState == null) { + /** + * Evaluates the indexed vector of a data row state. A row write evaluates each row state one + * time, and all checks of the write use the result. Returns null if the row state is null or has + * no vector, or if the index is not a vector index. If {@code strict} is true, a malformed vector + * causes an {@link IllegalDataException}. If it is false, a malformed vector gives null. + */ + public ImmutableBytesWritable getIndexedVector(Put dataRowState, boolean strict) { + if (dataRowState == null || !isVectorIndex()) { + return null; + } + return getIndexedVector(new IndexUtil.SimpleValueGetter(dataRowState), + HConstants.LATEST_TIMESTAMP, strict); + } + + /** + * Evaluates the indexed vector expression at a timestamp. Use strict evaluation for the new row + * state of a write, so that a malformed vector makes the write fail. Use lenient evaluation for + * the current row state. Then a malformed stored vector gives null and does not block a delete or + * a correction of the row. + */ + private ImmutableBytesWritable getIndexedVector(ValueGetter valueGetter, long ts, + boolean strict) { + if (valueGetter == 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 - ) { + try { + if ( + !indexedExpressions.get(0).evaluate(new ValueGetterTuple(valueGetter, ts), ptr) + || ptr.getLength() == 0 + ) { + return null; + } + } catch (IllegalDataException e) { + if (strict) { + throw e; + } 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. + * Returns true if the indexed vectors of the current and next row states have identical bytes. + * Then the next row keeps the centroid and the index row key of the current row. Returns false if + * either vector is null. */ - public boolean isVectorUnchanged(Put currentDataRowState, Put nextDataRowState) { - if (!isVectorIndex()) { - return false; - } - ImmutableBytesWritable current = getIndexedVector(currentDataRowState); - ImmutableBytesWritable next = getIndexedVector(nextDataRowState); + public boolean isVectorUnchanged(ImmutableBytesWritable current, ImmutableBytesWritable next) { return current != null && next != null && Bytes.equals(current.get(), current.getOffset(), current.getLength(), next.get(), next.getOffset(), next.getLength()); } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ExplainTable.java b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ExplainTable.java index 6135dce8b4..589b0437eb 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ExplainTable.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ExplainTable.java @@ -572,7 +572,8 @@ public abstract class ExplainTable { Arrays.asList(BaseScannerRegionObserverConstants.JSON_VALUE_FUNCTION, BaseScannerRegionObserverConstants.JSON_QUERY_FUNCTION)); emitServerProjection(planSteps, explainPlanAttributesBuilder, "BSON", - Collections.singletonList(BaseScannerRegionObserverConstants.BSON_VALUE_FUNCTION)); + Arrays.asList(BaseScannerRegionObserverConstants.BSON_VALUE_FUNCTION, + BaseScannerRegionObserverConstants.BSON_VECTOR_VALUE_FUNCTION)); } /** diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/OptimizerReasons.java b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/OptimizerReasons.java index cc84f798ac..2b44d05e54 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/OptimizerReasons.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/OptimizerReasons.java @@ -56,6 +56,8 @@ public final class OptimizerReasons { public static final String REASON_VECTOR_NULLS_FIRST = "NULLS FIRST needs rows the index does not hold"; public static final String REASON_VECTOR_COLUMN_MISMATCH = "indexes a different vector column"; + public static final String REASON_VECTOR_EXPRESSION_NOT_INDEXED = + "does not index the query vector expression"; /** Builds the functional index rule label of the form {@code "matches <expr>"}. */ public static String matches(String expression) { diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java index 13c76dad0f..b54584bc10 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java @@ -600,8 +600,11 @@ public class QueryOptimizer { targetColumns, parallelIteratorFactory, dataPlan.getContext().getSequenceManager(), isProjected, true, dataPlans).withRewriteContext(dataPlan.getContext()); - if (index.isVectorIndex() && !ordersByIndexedVector(index, rewrittenIndexSelect)) { - return AddPlanResult.rejected(index, OptimizerReasons.REASON_VECTOR_COLUMN_MISMATCH); + if (index.isVectorIndex()) { + String reason = getVectorOrderingRejection(index, rewrittenIndexSelect); + if (reason != null) { + return AddPlanResult.rejected(index, reason); + } } QueryPlan plan; if (index.isVectorIndex()) { @@ -830,11 +833,17 @@ public class QueryOptimizer { * Returns the reason to reject the index if the rewritten index query does not rank by the * indexed vector column of the index. Returns null if it does. */ - private static boolean ordersByIndexedVector(PTable index, SelectStatement indexSelect) { + private static String getVectorOrderingRejection(PTable index, SelectStatement indexSelect) { VectorSearchDescriptor descriptor = VectorSearchUtil.getVectorSearchDescriptor(indexSelect); + if (descriptor == null || !descriptor.isSourceColumn()) { + // The rewrite did not replace the ranked vector with an index column, so the index does not + // store this vector + return OptimizerReasons.REASON_VECTOR_EXPRESSION_NOT_INDEXED; + } PColumn vectorColumn = VectorIndexTrainer.getIndexedVectorColumn(index); - return descriptor != null && descriptor.isSourceColumn() && vectorColumn != null - && vectorColumn.getName().getString().equals(descriptor.getSourceColumnName()); + return vectorColumn.getName().getString().equals(descriptor.getSourceColumnName()) + ? null + : OptimizerReasons.REASON_VECTOR_COLUMN_MISMATCH; } /** diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/BsonVectorValueParseNode.java b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/BsonVectorValueParseNode.java new file mode 100644 index 0000000000..5c63a93886 --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/BsonVectorValueParseNode.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.parse; + +import java.sql.SQLException; +import java.util.List; +import org.apache.phoenix.compile.StatementContext; +import org.apache.phoenix.exception.SQLExceptionCode; +import org.apache.phoenix.exception.SQLExceptionInfo; +import org.apache.phoenix.expression.Expression; +import org.apache.phoenix.expression.function.BsonVectorValueFunction; +import org.apache.phoenix.expression.function.FunctionExpression; +import org.apache.phoenix.schema.types.PBson; +import org.apache.phoenix.schema.types.PDataType; +import org.apache.phoenix.schema.types.PJson; + +public class BsonVectorValueParseNode extends FunctionParseNode { + + public BsonVectorValueParseNode(String name, List<ParseNode> children, BuiltInFunctionInfo info) { + super(name, children, info); + } + + @Override + public FunctionExpression create(List<Expression> children, StatementContext context) + throws SQLException { + PDataType<?> dataType = children.get(0).getDataType(); + if (!dataType.isCoercibleTo(PJson.INSTANCE) && !dataType.isCoercibleTo(PBson.INSTANCE)) { + throw new SQLExceptionInfo.Builder(SQLExceptionCode.TYPE_MISMATCH) + .setMessage(dataType + " type is unsupported for BSON_VECTOR_VALUE()").build() + .buildException(); + } + try { + return new BsonVectorValueFunction(children); + } catch (IllegalArgumentException e) { + throw new SQLExceptionInfo.Builder(SQLExceptionCode.ILLEGAL_DATA).setMessage(e.getMessage()) + .build().buildException(); + } + } +} diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/IndexExpressionParseNodeRewriter.java b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/IndexExpressionParseNodeRewriter.java index b0214976b2..30a6869fb5 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/IndexExpressionParseNodeRewriter.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/IndexExpressionParseNodeRewriter.java @@ -28,6 +28,7 @@ import org.apache.phoenix.compile.FromCompiler; import org.apache.phoenix.compile.IndexStatementRewriter; import org.apache.phoenix.compile.StatementContext; import org.apache.phoenix.expression.Expression; +import org.apache.phoenix.index.vector.VectorIndexTrainer; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.jdbc.PhoenixStatement; import org.apache.phoenix.schema.PColumn; @@ -96,31 +97,46 @@ public class IndexExpressionParseNodeRewriter extends ParseNodeRewriter { // The index calculates the centroid ID from the vector, so no data expression maps to it continue; } - String expressionStr = IndexUtil.getIndexColumnExpressionStr(column); - ParseNode expressionParseNode = SQLParser.parseCondition(expressionStr); - String colName = "\"" + column.getName().getString() + "\""; - Expression dataExpression = expressionParseNode.accept(expressionCompiler); - PDataType expressionDataType = dataExpression.getDataType(); - ParseNode indexedParseNode = expressionParseNode.accept(rewriter); - PDataType indexColType = - IndexUtil.getIndexColumnDataType(dataExpression.isNullable(), expressionDataType); - ParseNode columnParseNode = - new ColumnParseNode(alias != null ? TableName.create(null, alias) : null, colName, null); - if (indexColType != expressionDataType) { - columnParseNode = NODE_FACTORY.cast(columnParseNode, expressionDataType, null, null); - } - indexedParseNodeToColumnParseNodeMap.put(indexedParseNode, columnParseNode); - // Only true functional columns defined over an expression get an applied match entry. A - // plain indexed or PK column's expression string parses to a bare column reference. An - // expression column (e.g. UPPER(NAME)) parses to a compound node. - if (!(expressionParseNode instanceof ColumnParseNode)) { - // Trim leading/trailing whitespace - indexedParseNodeToFunctionalColumn.put(indexedParseNode, - new String[] { colName, expressionStr.trim() }); + addMapping(column, alias, expressionCompiler, rewriter); + } + if (index.isVectorIndex()) { + // A functional vector index keeps the computed vector in a non-PK index column. Map the + // vector expression to that column, so that a query that ranks by the expression can use the + // index. + PColumn vectorColumn = VectorIndexTrainer.getIndexedVectorColumn(index); + if (!(SQLParser.parseCondition(vectorColumn.getExpressionStr()) instanceof ColumnParseNode)) { + addMapping(vectorColumn, alias, expressionCompiler, rewriter); } } } + /** Registers the rewrite from the indexed expression of an index column to that column. */ + private void addMapping(PColumn column, String alias, ExpressionCompiler expressionCompiler, + IndexStatementRewriter rewriter) throws SQLException { + String expressionStr = IndexUtil.getIndexColumnExpressionStr(column); + ParseNode expressionParseNode = SQLParser.parseCondition(expressionStr); + String colName = "\"" + column.getName().getString() + "\""; + Expression dataExpression = expressionParseNode.accept(expressionCompiler); + PDataType expressionDataType = dataExpression.getDataType(); + ParseNode indexedParseNode = expressionParseNode.accept(rewriter); + PDataType indexColType = + IndexUtil.getIndexColumnDataType(dataExpression.isNullable(), expressionDataType); + ParseNode columnParseNode = + new ColumnParseNode(alias != null ? TableName.create(null, alias) : null, colName, null); + if (indexColType != expressionDataType) { + columnParseNode = NODE_FACTORY.cast(columnParseNode, expressionDataType, null, null); + } + indexedParseNodeToColumnParseNodeMap.put(indexedParseNode, columnParseNode); + // Only a functional column, which is defined over an expression, gets an applied match entry. + // The expression string of a plain indexed or PK column parses to a bare column reference. The + // expression string of a functional column, for example UPPER(NAME), parses to a compound node. + if (!(expressionParseNode instanceof ColumnParseNode)) { + // Remove the leading and trailing whitespace + indexedParseNodeToFunctionalColumn.put(indexedParseNode, + new String[] { colName, expressionStr.trim() }); + } + } + @Override protected ParseNode leaveCompoundNode(CompoundParseNode node, List<ParseNode> children, CompoundNodeFactory factory) { diff --git a/phoenix-core-client/src/main/protobuf/ServerCachingService.proto b/phoenix-core-client/src/main/protobuf/ServerCachingService.proto index 58911b6833..1c719fcc2f 100644 --- a/phoenix-core-client/src/main/protobuf/ServerCachingService.proto +++ b/phoenix-core-client/src/main/protobuf/ServerCachingService.proto @@ -77,6 +77,7 @@ message IndexMaintainer { optional string vectorAlgorithm = 34; optional string distanceMetric = 35; optional int64 centroidGeneration = 36; + optional ColumnReference functionalVectorColumn = 37; } 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 afa016154a..70ed585a92 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 @@ -1391,12 +1391,20 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver { for (Pair<IndexMaintainer, HTableInterfaceReference> pair : indexTables) { IndexMaintainer indexMaintainer = pair.getFirst(); HTableInterfaceReference hTableInterfaceReference = pair.getSecond(); + // Evaluate the indexed vector of each row state one time, because a BSON vector costs a + // document decode. The new state is strict, so a malformed vector makes the write fail. The + // current state is lenient, so a malformed stored vector does not block a delete or a + // correction. + ImmutableBytesWritable nextVector = indexMaintainer.getIndexedVector(nextDataRowState, true); + ImmutableBytesWritable currentVector = + indexMaintainer.getIndexedVector(currentDataRowState, false); if ( - nextDataRowState != null && indexMaintainer.shouldPrepareIndexMutations(nextDataRowState) + nextDataRowState != null + && indexMaintainer.shouldPrepareIndexMutations(nextDataRowState, nextVector) ) { ValueGetter nextDataRowVG = new IndexUtil.SimpleValueGetter(nextDataRowState); Put indexPut = indexMaintainer.buildUpdateMutation(GenericKeyValueBuilder.INSTANCE, - nextDataRowVG, rowKeyPtr, ts, null, null, false, encodedRegionName); + nextDataRowVG, rowKeyPtr, ts, null, null, false, encodedRegionName, nextVector); if (indexPut == null) { // No covered column. Just prepare an index row with the empty column byte[] indexRowKey = indexMaintainer.buildRowKey(nextDataRowVG, rowKeyPtr, null, null, ts, @@ -1422,12 +1430,12 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver { // and the index is not CDC. Skip centroid assignment when the indexed vector is unchanged. if ( currentDataRowState != null - && !indexMaintainer.isVectorUnchanged(currentDataRowState, nextDataRowState) + && !indexMaintainer.isVectorUnchanged(currentVector, nextVector) ) { 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); + rowKeyPtr, ts, encodedRegionName, currentVector); if ( indexRowKeyForCurrentDataRow != null && !indexMaintainer.isCDCIndex() && Bytes.compareTo(indexPut.getRow(), indexRowKeyForCurrentDataRow) != 0 @@ -1439,7 +1447,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver { } } else if ( currentDataRowState != null - && indexMaintainer.shouldPrepareIndexMutations(currentDataRowState) + && indexMaintainer.shouldPrepareIndexMutations(currentDataRowState, currentVector) ) { if (indexMaintainer.isCDCIndex()) { // CDC Index needs two a delete marker for referencing the data table @@ -1451,8 +1459,11 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver { indexUpdates.put(hTableInterfaceReference, getDeleteIndexMutation(cdcDataRowState, indexMaintainer, ts, rowKeyPtr, encodedRegionName)); } else { - indexUpdates.put(hTableInterfaceReference, getDeleteIndexMutation(currentDataRowState, - indexMaintainer, ts, rowKeyPtr, encodedRegionName)); + Mutation del = indexMaintainer.buildRowDeleteMutation( + indexMaintainer.buildRowKey(new IndexUtil.SimpleValueGetter(currentDataRowState), + rowKeyPtr, ts, encodedRegionName, currentVector), + IndexMaintainer.DeleteType.ALL_VERSIONS, ts); + indexUpdates.put(hTableInterfaceReference, del); } } } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/iterate/NonAggregateRegionScannerFactory.java b/phoenix-core-server/src/main/java/org/apache/phoenix/iterate/NonAggregateRegionScannerFactory.java index dd1db9f0ec..12730f82a6 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/iterate/NonAggregateRegionScannerFactory.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/iterate/NonAggregateRegionScannerFactory.java @@ -57,6 +57,7 @@ import org.apache.phoenix.expression.OrderByExpression; import org.apache.phoenix.expression.SingleCellColumnExpression; import org.apache.phoenix.expression.function.ArrayIndexFunction; import org.apache.phoenix.expression.function.BsonValueFunction; +import org.apache.phoenix.expression.function.BsonVectorValueFunction; import org.apache.phoenix.expression.function.JsonQueryFunction; import org.apache.phoenix.expression.function.JsonValueFunction; import org.apache.phoenix.expression.function.ScalarFunction; @@ -231,6 +232,9 @@ public class NonAggregateRegionScannerFactory extends RegionScannerFactory { if (serverParsedJsonQueryFuncRefs != null) { Collections.addAll(resultList, serverParsedJsonQueryFuncRefs); } + deserializeAndAddComplexDataTypeFunctions(scan, + BaseScannerRegionObserverConstants.BSON_VECTOR_VALUE_FUNCTION, serverParsedKVRefs, + resultList); return resultList; } @@ -357,7 +361,10 @@ public class NonAggregateRegionScannerFactory extends RegionScannerFactory { func = new JsonQueryFunction(); } else if (scanAttribute.equals(BaseScannerRegionObserverConstants.BSON_VALUE_FUNCTION)) { func = new BsonValueFunction(); - } + } else + if (scanAttribute.equals(BaseScannerRegionObserverConstants.BSON_VECTOR_VALUE_FUNCTION)) { + func = new BsonVectorValueFunction(); + } if (func != null) { func.readFields(input); funcRefs[i] = func; diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java b/phoenix-core-server/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java index 1cbb6f67d3..0ab1b0494c 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java @@ -61,6 +61,7 @@ import org.apache.phoenix.schema.PColumn; import org.apache.phoenix.schema.PColumnImpl; import org.apache.phoenix.schema.PTable; import org.apache.phoenix.schema.ValueBitSet; +import org.apache.phoenix.schema.tuple.EncodedColumnQualiferCellsList; import org.apache.phoenix.schema.tuple.MultiKeyValueTuple; import org.apache.phoenix.schema.tuple.PositionBasedResultTuple; import org.apache.phoenix.schema.tuple.ResultTuple; @@ -473,8 +474,24 @@ public abstract class RegionScannerFactory { }; } - // PHOENIX-4791 Share position of array element cell + // PHOENIX-4791 Returns the position of the array element cell, which holds the server parsed + // projection values. An ordinary list has this cell at the end. An encoded qualifier list keeps + // the cell in its reserved qualifier range, after the empty column cell and before the other + // column cells. Thus this method finds the cell by its column. The method iterates because + // indexed access to an encoded list scans the list from the start. public static int getArrayCellPosition(List<Cell> result) { + if (result instanceof EncodedColumnQualiferCellsList) { + int i = 0; + for (Cell cell : result) { + if ( + CellUtil.matchingColumn(cell, QueryConstants.ARRAY_VALUE_COLUMN_FAMILY, + QueryConstants.ARRAY_VALUE_COLUMN_QUALIFIER) + ) { + return i; + } + i++; + } + } return result.size() - 1; } } diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BsonServerProjectionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BsonServerProjectionIT.java new file mode 100644 index 0000000000..cab7dc4106 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BsonServerProjectionIT.java @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.end2end; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.util.ArrayList; +import java.util.List; +import org.apache.phoenix.util.QueryUtil; +import org.bson.BsonDocument; +import org.bson.BsonInt32; +import org.bson.BsonString; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Integration tests for server side BSON projection together with a server top-N, and together with + * server parsed functions of different types. + */ +@Category(ParallelStatsDisabledTest.class) +public class BsonServerProjectionIT extends ParallelStatsDisabledIT { + + private static String createTable(Connection conn) throws Exception { + return createTable(conn, ""); + } + + private static String createTable(Connection conn, String options) throws Exception { + String table = generateUniqueName(); + conn.createStatement().execute( + "CREATE TABLE " + table + " (ID VARCHAR PRIMARY KEY, N INTEGER, DOC BSON) " + options); + try ( + PreparedStatement ps = conn.prepareStatement("UPSERT INTO " + table + " VALUES (?, ?, ?)")) { + for (int i = 0; i < 5; i++) { + BsonDocument doc = new BsonDocument("name", new BsonString("name" + i)); + doc.put("rank", new BsonInt32(10 - i)); + doc.put("tag", new BsonDocument("k", new BsonString("tag" + i))); + ps.setString(1, "id" + i); + ps.setInt(2, i); + ps.setObject(3, doc); + ps.executeUpdate(); + } + } + conn.commit(); + return table; + } + + private static List<String> rows(Connection conn, String sql) throws Exception { + List<String> out = new ArrayList<>(); + try (ResultSet rs = conn.createStatement().executeQuery(sql)) { + int n = rs.getMetaData().getColumnCount(); + while (rs.next()) { + StringBuilder row = new StringBuilder(); + for (int i = 1; i <= n; i++) { + row.append(i > 1 ? "," : "").append(rs.getString(i)); + } + out.add(row.toString()); + } + } + return out; + } + + /** Tests server side BSON projection together with a server top-N (ORDER BY with LIMIT). */ + @Test + public void testProjectionWithServerTopN() throws Exception { + try (Connection conn = DriverManager.getConnection(getUrl())) { + String table = createTable(conn); + String sql = + "SELECT ID, BSON_VALUE(DOC, 'name', 'VARCHAR') FROM " + table + " ORDER BY N DESC LIMIT 2"; + assertTrue(QueryUtil.getExplainPlan(conn.createStatement().executeQuery("EXPLAIN " + sql)) + .contains("SERVER TOP 2")); + List<String> expected = new ArrayList<>(); + expected.add("id4,name4"); + expected.add("id3,name3"); + assertEquals(expected, rows(conn, sql)); + } + } + + /** Tests the same query on a table without column qualifier encoding. */ + @Test + public void testProjectionWithServerTopNWithoutColumnEncoding() throws Exception { + try (Connection conn = DriverManager.getConnection(getUrl())) { + String table = createTable(conn, "COLUMN_ENCODED_BYTES=0"); + String sql = + "SELECT ID, BSON_VALUE(DOC, 'name', 'VARCHAR') FROM " + table + " ORDER BY N DESC LIMIT 2"; + List<String> expected = new ArrayList<>(); + expected.add("id4,name4"); + expected.add("id3,name3"); + assertEquals(expected, rows(conn, sql)); + } + } + + /** + * Verifies an ORDER BY on the same BSON document column as a server parsed projection. The server + * must keep the document cell for the sort. + */ + @Test + public void testOrderByReadsProjectedDocument() throws Exception { + try (Connection conn = DriverManager.getConnection(getUrl())) { + String table = createTable(conn); + String sql = "SELECT ID, BSON_VALUE(DOC, 'name', 'VARCHAR') FROM " + table + + " ORDER BY BSON_VALUE(DOC, 'rank', 'INTEGER') LIMIT 2"; + List<String> expected = new ArrayList<>(); + expected.add("id4,name4"); + expected.add("id3,name3"); + assertEquals(expected, rows(conn, sql)); + } + } + + /** + * Verifies the decode order of server parsed functions of different types in one projection. The + * query lists them in an order different from the server evaluation order (array element, JSON + * value, BSON value, JSON query). A client layout that does not match the server layout reads + * wrong values. + */ + @Test + public void testMixedFunctionProjection() throws Exception { + try (Connection conn = DriverManager.getConnection(getUrl())) { + String table = generateUniqueName(); + conn.createStatement().execute( + "CREATE TABLE " + table + " (ID VARCHAR PRIMARY KEY, A INTEGER ARRAY, J JSON, DOC BSON)"); + BsonDocument doc = new BsonDocument("name", new BsonString("bson-name")); + doc.put("rank", new BsonInt32(8)); + try (PreparedStatement ps = + conn.prepareStatement("UPSERT INTO " + table + " VALUES (?, ?, ?, ?)")) { + ps.setString(1, "id1"); + ps.setArray(2, conn.createArrayOf("INTEGER", new Integer[] { 7, 9 })); + ps.setString(3, "{\"k\": \"json-value\", \"obj\": {\"a\": 1}}"); + ps.setObject(4, doc); + ps.executeUpdate(); + } + conn.commit(); + String sql = "SELECT ID, JSON_QUERY(J, '$.obj'), BSON_VALUE(DOC, 'name', 'VARCHAR')," + + " JSON_VALUE(J, '$.k'), A[2], BSON_VALUE(DOC, 'rank', 'INTEGER') FROM " + table; + String plan = QueryUtil.getExplainPlan(conn.createStatement().executeQuery("EXPLAIN " + sql)); + assertTrue(plan, plan.contains("SERVER ARRAY PROJECTION 1")); + assertTrue(plan, plan.contains("SERVER JSON PROJECTION 2")); + assertTrue(plan, plan.contains("SERVER BSON PROJECTION 2")); + try (ResultSet rs = conn.createStatement().executeQuery(sql)) { + assertTrue(rs.next()); + assertEquals("id1", rs.getString(1)); + assertEquals("{\"a\":1}", rs.getString(2).replaceAll("\\s", "")); + assertEquals("bson-name", rs.getString(3)); + assertEquals("json-value", rs.getString(4)); + assertEquals(9, rs.getInt(5)); + assertEquals(8, rs.getInt(6)); + assertFalse(rs.next()); + } + } + } +} diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorBsonIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorBsonIndexIT.java new file mode 100644 index 0000000000..c5bce0b891 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorBsonIndexIT.java @@ -0,0 +1,589 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.end2end; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.phoenix.hbase.index.covered.update.ColumnReference; +import org.apache.phoenix.index.IndexMaintainer; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.types.PVarchar; +import org.apache.phoenix.schema.types.PVectorFloat; +import org.apache.phoenix.util.QueryUtil; +import org.bson.BinaryVector; +import org.bson.BsonBinary; +import org.bson.BsonBinarySubType; +import org.bson.BsonDocument; +import org.bson.BsonNull; +import org.bson.BsonString; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Integration tests for BSON_VECTOR_VALUE: vector extraction from BSON documents, functional vector + * indexes, index selection by the optimizer, and server side projection. + */ +@Category(ParallelStatsDisabledTest.class) +public class VectorBsonIndexIT extends ParallelStatsDisabledIT { + + private static final int DIM = 8; + + private static BsonBinary bsonVector(float[] v) { + return new BsonBinary(BinaryVector.floatVector(v)); + } + + /** Returns a test vector with the first component {@code x} and all other components zero. */ + private static float[] vecX(float x) { + float[] v = new float[DIM]; + v[0] = x; + return v; + } + + private static BsonDocument embeddingDoc(float[] v, String category) { + BsonDocument doc = new BsonDocument("embedding", bsonVector(v)); + doc.put("category", new BsonString(category)); + return doc; + } + + private static void upsertDoc(Connection conn, String table, String id, BsonDocument doc) + throws SQLException { + try (PreparedStatement ps = conn.prepareStatement("UPSERT INTO " + table + " VALUES (?, ?)")) { + ps.setString(1, id); + ps.setObject(2, doc); + ps.executeUpdate(); + } + conn.commit(); + } + + private static int countRows(Connection conn, String table) throws SQLException { + try (Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM " + table)) { + assertTrue(rs.next()); + return rs.getInt(1); + } + } + + private static Float[] boxed(float[] v) { + Float[] b = new Float[v.length]; + for (int i = 0; i < v.length; i++) { + b[i] = v[i]; + } + return b; + } + + private static List<String> runSearch(Connection conn, String sql, float[] q) + throws SQLException { + List<String> ids = new ArrayList<>(); + try (PreparedStatement ps = conn.prepareStatement(sql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxed(q))); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + ids.add(rs.getString(1)); + } + } + } + return ids; + } + + private static String explain(Connection conn, String sql, float[] q) throws SQLException { + try (PreparedStatement ps = conn.prepareStatement("EXPLAIN " + sql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxed(q))); + return QueryUtil.getExplainPlan(ps.executeQuery()); + } + } + + /** Returns four test centroids at 0, 10, 20, and 30 on the first dimension. */ + private static List<float[]> fourCentroids() { + List<float[]> centroids = new ArrayList<>(); + for (int c = 0; c < 4; c++) { + centroids.add(vecX(c * 10.0f)); + } + return centroids; + } + + /** Returns the centroid ID of each index row, keyed by the data row ID. */ + private static Map<String, Integer> indexCentroidById(PhoenixConnection pconn, String indexName) + throws Exception { + PTable indexTable = pconn.getTableNoCache(indexName); + Map<String, Integer> byId = new HashMap<>(); + for (byte[] rk : VectorIndexTestUtil.getHBaseRowKeys(pconn, indexTable)) { + String id = + (String) PVarchar.INSTANCE.toObject(rk, Bytes.SIZEOF_INT, rk.length - Bytes.SIZEOF_INT); + byId.put(id, VectorIndexTestUtil.extractCentroidId(rk)); + } + return byId; + } + + private static void createIndexedTable(Connection conn, String table, String index) + throws SQLException { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + table + " (ID VARCHAR NOT NULL PRIMARY KEY, DOC BSON)"); + stmt.execute( + "CREATE VECTOR INDEX " + index + " ON " + table + " (BSON_VECTOR_VALUE(doc, 'embedding', " + + DIM + ")) WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)"); + } + } + + @Test + public void testBsonFunctionalVectorIndexWriteLifecycle() throws Exception { + String tableName = "T_BSON_FUNC_" + generateUniqueName(); + String indexName = "IDX_BSON_FUNC_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createIndexedTable(conn, tableName, indexName); + List<float[]> centroids = fourCentroids(); + VectorIndexTestUtil.activateWithKnownCentroids(conn, tableName, indexName, centroids); + PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); + + // A functional vector index has an index column for the computed vector. An index on a plain + // vector column has no such column. + PTable dataTable = pconn.getTableNoCache(tableName); + PTable indexTable = pconn.getTableNoCache(indexName); + IndexMaintainer maintainer = indexTable.getIndexMaintainer(dataTable, pconn); + assertNotNull(maintainer.getFunctionalVectorColumn()); + String colTable = "T_COL_VEC_" + generateUniqueName(); + String colIndex = "IDX_COL_VEC_" + generateUniqueName(); + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + colTable + + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, " + DIM + "))"); + stmt.execute("CREATE VECTOR INDEX " + colIndex + " ON " + colTable + + " (V) WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100)"); + } + assertNull(pconn.getTableNoCache(colIndex) + .getIndexMaintainer(pconn.getTableNoCache(colTable), pconn).getFunctionalVectorColumn()); + + // Write rows near all centroids, rows without a vector field, and a row with a null vector + Map<String, float[]> vectors = new LinkedHashMap<>(); + for (int i = 0; i < 40; i++) { + float[] v = vecX(i); + vectors.put("id_" + i, v); + upsertDoc(conn, tableName, "id_" + i, embeddingDoc(v, "c" + (i % 3))); + } + for (int i = 0; i < 5; i++) { + upsertDoc(conn, tableName, "noemb_" + i, + new BsonDocument("category", new BsonString("none"))); + } + upsertDoc(conn, tableName, "nullemb", new BsonDocument("embedding", BsonNull.VALUE)); + assertEquals(46, countRows(conn, tableName)); + assertEquals(40, countRows(conn, indexName)); + + Map<String, Integer> byId = indexCentroidById(pconn, indexName); + assertEquals(40, byId.size()); + for (Map.Entry<String, float[]> e : vectors.entrySet()) { + int expected = VectorIndexTestUtil.nearestCentroid(e.getValue(), centroids, "L2"); + assertEquals("centroid of " + e.getKey(), Integer.valueOf(expected), byId.get(e.getKey())); + } + + // The index row must hold the vector in the little endian PVectorFloat encoding + ColumnReference vecRef = maintainer.getFunctionalVectorColumn(); + byte[] rowKeyOf7 = null; + for (byte[] rk : VectorIndexTestUtil.getHBaseRowKeys(pconn, indexTable)) { + String id = + (String) PVarchar.INSTANCE.toObject(rk, Bytes.SIZEOF_INT, rk.length - Bytes.SIZEOF_INT); + if ("id_7".equals(id)) { + rowKeyOf7 = rk; + } + } + assertNotNull(rowKeyOf7); + try ( + Table hTable = pconn.getQueryServices().getTable(indexTable.getPhysicalName().getBytes())) { + Result r = hTable.get(new Get(rowKeyOf7)); + assertArrayEquals(PVectorFloat.INSTANCE.toBytes(vecX(7)), + r.getValue(vecRef.getFamily(), vecRef.getQualifier())); + } + + // A change to a field other than the vector keeps the index row key + upsertDoc(conn, tableName, "id_7", embeddingDoc(vecX(7), "renamed")); + assertEquals(40, countRows(conn, indexName)); + assertEquals(byId.get("id_7"), indexCentroidById(pconn, indexName).get("id_7")); + + // A change to the vector moves the index row to the new nearest centroid + assertEquals(Integer.valueOf(1), byId.get("id_7")); + upsertDoc(conn, tableName, "id_7", embeddingDoc(vecX(35.5f), "renamed")); + vectors.put("id_7", vecX(35.5f)); + assertEquals(40, countRows(conn, indexName)); + assertEquals(Integer.valueOf(3), indexCentroidById(pconn, indexName).get("id_7")); + + // A document without the vector field removes the index row + upsertDoc(conn, tableName, "id_8", new BsonDocument("category", new BsonString("gone"))); + vectors.remove("id_8"); + assertEquals(46, countRows(conn, tableName)); + assertEquals(39, countRows(conn, indexName)); + assertFalse(indexCentroidById(pconn, indexName).containsKey("id_8")); + + // A malformed vector payload makes the write fail + BsonBinary[] malformed = new BsonBinary[] { bsonVector(new float[DIM + 1]), + new BsonBinary(BinaryVector.int8Vector(new byte[DIM])), + new BsonBinary(BsonBinarySubType.BINARY, bsonVector(vecX(1)).getData()) }; + for (BsonBinary bad : malformed) { + try { + upsertDoc(conn, tableName, "bad", new BsonDocument("embedding", bad)); + fail("Malformed embedding " + bad + " must be rejected at write time"); + } catch (SQLException expected) { + conn.rollback(); + } + } + assertEquals(46, countRows(conn, tableName)); + assertEquals(39, countRows(conn, indexName)); + + float[] q = vecX(35.3f); + String sql = "SELECT ID FROM " + tableName + " ORDER BY L2_DISTANCE(BSON_VECTOR_VALUE(doc, " + + "'embedding', " + DIM + "), ?) LIMIT 3"; + assertTrue(explain(conn, sql, q).contains(indexName)); + assertEquals(VectorIndexTestUtil.bruteForceTopK(vectors, q, "L2", 3), + runSearch(conn, sql, q)); + } + } + + /** + * Verifies that a stored row with a malformed vector and no index row does not block a delete or + * a correction of that row. A new write of a malformed vector must still fail. + */ + @Test + public void testMalformedCurrentVectorDoesNotBlockDeleteOrCorrection() throws Exception { + String tableName = "T_BSON_BAD_" + generateUniqueName(); + String indexName = "IDX_BSON_BAD_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + // An untrained index has no index maintenance, so the table accepts these malformed rows + createIndexedTable(conn, tableName, indexName); + BsonDocument bad = + new BsonDocument("embedding", new BsonBinary(BinaryVector.int8Vector(new byte[DIM]))); + float[] nanVector = new float[DIM]; + nanVector[DIM - 1] = Float.NaN; + BsonDocument nan = + new BsonDocument("embedding", new BsonBinary(BinaryVector.floatVector(nanVector))); + upsertDoc(conn, tableName, "fix", bad); + upsertDoc(conn, tableName, "del", nan); + upsertDoc(conn, tableName, "keep", bad); + VectorIndexTestUtil.activateWithKnownCentroids(conn, tableName, indexName, fourCentroids()); + assertEquals(0, countRows(conn, indexName)); + + upsertDoc(conn, tableName, "fix", embeddingDoc(vecX(21f), "fixed")); + conn.createStatement().execute("DELETE FROM " + tableName + " WHERE ID = 'del'"); + conn.commit(); + for (BsonDocument malformed : new BsonDocument[] { bad, nan }) { + try { + upsertDoc(conn, tableName, "keep", malformed); + fail("Writing a malformed vector must fail"); + } catch (SQLException expected) { + conn.rollback(); + } + } + assertEquals(2, countRows(conn, tableName)); + PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); + Map<String, Integer> byId = indexCentroidById(pconn, indexName); + assertEquals(1, byId.size()); + assertEquals(Integer.valueOf(2), byId.get("fix")); + } + } + + @Test + public void testBsonFunctionalVectorIndexSearch() throws Exception { + String tableName = "T_BSON_QUERY_" + generateUniqueName(); + String indexName = "IDX_BSON_QUERY_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + createIndexedTable(conn, tableName, indexName); + VectorIndexTestUtil.activateWithKnownCentroids(conn, tableName, indexName, fourCentroids()); + + // Write vectors on both sides of each boundary between centroids + Map<String, float[]> vectors = new LinkedHashMap<>(); + float[] xs = { 1f, 4f, 6f, 9f, 11f, 14f, 16f, 19f, 21f, 24f, 26f, 29f }; + for (int i = 0; i < xs.length; i++) { + float[] v = vecX(xs[i]); + v[1] = 0.1f * i; + vectors.put("r" + i, v); + upsertDoc(conn, tableName, "r" + i, embeddingDoc(v, "c" + (i % 2))); + } + + String sql = "SELECT ID FROM " + tableName + " ORDER BY L2_DISTANCE(BSON_VECTOR_VALUE(doc, " + + "'embedding', " + DIM + "), ?) LIMIT 3"; + for (float qx : new float[] { 5f, 15f, 25f, 0f }) { + float[] q = vecX(qx); + String plan = explain(conn, sql, q); + assertTrue("Plan must use the BSON functional vector index: " + plan, + plan.contains(indexName) && plan.contains("CLIENT PROBING")); + assertEquals("query x=" + qx, VectorIndexTestUtil.bruteForceTopK(vectors, q, "L2", 3), + runSearch(conn, sql, q)); + } + + // The query vector can be the first distance operand. The plan must still use the index. + String flipped = "SELECT ID FROM " + tableName + + " ORDER BY L2_DISTANCE(?, BSON_VECTOR_VALUE(doc, 'embedding', " + DIM + ")) LIMIT 3"; + float[] q = vecX(15f); + assertTrue(explain(conn, flipped, q).contains(indexName)); + assertEquals(VectorIndexTestUtil.bruteForceTopK(vectors, q, "L2", 3), + runSearch(conn, flipped, q)); + } + } + + /** + * Verifies a functional vector index that also covers a vector column of another dimension. The + * covered column must not change the index dimension or the centroid assignment. + */ + @Test + public void testBsonVectorIndexWithCoveredVectorColumnOfAnotherDimension() throws Exception { + String tableName = "T_BSON_COV_" + generateUniqueName(); + String indexName = "IDX_BSON_COV_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + tableName + + " (ID VARCHAR NOT NULL PRIMARY KEY, DOC BSON, V4 VECTOR(FLOAT, 4))"); + stmt.execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + + " (BSON_VECTOR_VALUE(doc, 'embedding', " + DIM + ")) INCLUDE (V4)" + + " WITH (algorithm = 'IVF', metric = 'L2', lists = 2, sample_size = 100)"); + } + PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); + PTable indexTable = pconn.getTableNoCache(indexName); + assertEquals(Integer.valueOf(DIM), indexTable.getVectorDimension()); + assertNotNull(indexTable.getIndexMaintainer(pconn.getTableNoCache(tableName), pconn) + .getFunctionalVectorColumn()); + + VectorIndexTestUtil.activateWithKnownCentroids(conn, tableName, indexName, + Arrays.asList(vecX(0f), vecX(10f))); + try (PreparedStatement ps = + conn.prepareStatement("UPSERT INTO " + tableName + " VALUES (?, ?, ?)")) { + ps.setString(1, "r1"); + ps.setObject(2, new BsonDocument("embedding", bsonVector(vecX(11f)))); + ps.setArray(3, conn.createArrayOf("FLOAT", new Float[] { 1f, 2f, 3f, 4f })); + ps.executeUpdate(); + } + conn.commit(); + assertEquals(1, countRows(conn, indexName)); + assertEquals(Integer.valueOf(1), indexCentroidById(pconn, indexName).get("r1")); + } + } + + /** + * Verifies that the optimizer does not use the index if the query ranks by another document path, + * another dimension, or another vector column. + */ + @Test + public void testBsonVectorIndexNotUsedForDifferentExpression() throws Exception { + String tableName = "T_BSON_OTHER_" + generateUniqueName(); + String indexName = "IDX_BSON_OTHER_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + tableName + + " (ID VARCHAR NOT NULL PRIMARY KEY, DOC BSON, V VECTOR(FLOAT, " + DIM + "))"); + stmt.execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + + " (BSON_VECTOR_VALUE(doc, 'embedding', " + DIM + ")) INCLUDE (DOC, V)" + + " WITH (algorithm = 'IVF', metric = 'L2', lists = 2, sample_size = 100)"); + } + VectorIndexTestUtil.activateWithKnownCentroids(conn, tableName, indexName, + Arrays.asList(vecX(0f), vecX(10f))); + + String[] ids = { "A1", "A2", "A3", "B1", "B2", "B3" }; + float[] embX = { 2f, 3f, 4f, 6f, 7f, 8f }; + float[] otherX = { 100f, 101f, 102f, 0f, 1f, 2f }; + Map<String, float[]> otherRows = new LinkedHashMap<>(); + try (PreparedStatement ps = + conn.prepareStatement("UPSERT INTO " + tableName + " VALUES (?, ?, ?)")) { + for (int i = 0; i < ids.length; i++) { + BsonDocument doc = new BsonDocument("embedding", bsonVector(vecX(embX[i]))); + doc.put("other", bsonVector(vecX(otherX[i]))); + ps.setString(1, ids[i]); + ps.setObject(2, doc); + ps.setArray(3, conn.createArrayOf("FLOAT", boxed(vecX(otherX[i])))); + ps.executeUpdate(); + otherRows.put(ids[i], vecX(otherX[i])); + } + conn.commit(); + } + + float[] q = vecX(0f); + List<String> expected = VectorIndexTestUtil.bruteForceTopK(otherRows, q, "L2", 2); + String[] sqls = { + "SELECT ID FROM " + tableName + " ORDER BY L2_DISTANCE(BSON_VECTOR_VALUE(doc, 'other', " + + DIM + "), ?) LIMIT 2", + "SELECT ID FROM " + tableName + " ORDER BY L2_DISTANCE(V, ?) LIMIT 2" }; + for (String sql : sqls) { + String plan = explain(conn, sql, q); + assertFalse("Index on 'embedding' must not serve: " + sql + "\n" + plan, + plan.contains(indexName)); + assertEquals(sql, expected, runSearch(conn, sql, q)); + } + + String dimSql = + "SELECT ID FROM " + tableName + " ORDER BY L2_DISTANCE(BSON_VECTOR_VALUE(doc, 'embedding', " + + (DIM + 1) + "), ?) LIMIT 2"; + assertFalse(explain(conn, dimSql, new float[DIM + 1]).contains(indexName)); + + Map<String, float[]> embRows = new LinkedHashMap<>(); + for (int i = 0; i < ids.length; i++) { + embRows.put(ids[i], vecX(embX[i])); + } + String embSql = "SELECT ID FROM " + tableName + + " ORDER BY L2_DISTANCE(BSON_VECTOR_VALUE(doc, 'embedding', " + DIM + "), ?) LIMIT 2"; + assertTrue(explain(conn, embSql, q).contains(indexName)); + assertEquals(VectorIndexTestUtil.bruteForceTopK(embRows, q, "L2", 2), + runSearch(conn, embSql, q)); + } + } + + /** + * Tests exact nearest neighbor search over document vectors without a vector index, also with a + * BSON_VALUE projection in the query. + */ + @Test + public void testExactSearchOverDocumentVectors() throws Exception { + String tableName = "T_BSON_EXACT_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.createStatement() + .execute("CREATE TABLE " + tableName + " (ID VARCHAR NOT NULL PRIMARY KEY, DOC BSON)"); + Map<String, float[]> vectors = new LinkedHashMap<>(); + for (int i = 0; i < 20; i++) { + float[] v = vecX((i * 7) % 20); + v[1] = i; + vectors.put("r" + i, v); + upsertDoc(conn, tableName, "r" + i, embeddingDoc(v, "n" + i)); + } + float[] q = vecX(9.5f); + q[1] = 3f; + List<String> expected = VectorIndexTestUtil.bruteForceTopK(vectors, q, "L2", 5); + String distance = "L2_DISTANCE(BSON_VECTOR_VALUE(doc, 'embedding', " + DIM + "), ?)"; + String[] sqls = { "SELECT ID FROM " + tableName + " ORDER BY " + distance + " LIMIT 5", + "SELECT ID, " + distance + " D FROM " + tableName + " ORDER BY D LIMIT 5", + "SELECT ID, BSON_VALUE(DOC, 'category', 'VARCHAR') FROM " + tableName + " ORDER BY " + + distance + " LIMIT 5" }; + for (String sql : sqls) { + assertEquals(sql, expected, runSearch(conn, sql, q)); + } + try (PreparedStatement ps = conn.prepareStatement(sqls[2])) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxed(q))); + try (ResultSet rs = ps.executeQuery()) { + for (String id : expected) { + assertTrue(rs.next()); + assertEquals(id, rs.getString(1)); + assertEquals("n" + id.substring(1), rs.getString(2)); + } + } + } + } + } + + @Test + public void testServerSideBsonVectorProjection() throws Exception { + String tableName = "T_BSON_PROJ_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.createStatement() + .execute("CREATE TABLE " + tableName + " (ID VARCHAR NOT NULL PRIMARY KEY, DOC BSON)"); + float[] v1 = new float[] { 1.5f, -2.5f, 3.0e-3f }; + float[] v2 = new float[] { 4.0f, 5.0f, 6.0f }; + BsonDocument data1 = new BsonDocument("vec", bsonVector(v1)); + data1.put("name", new BsonString("one")); + BsonDocument data2 = new BsonDocument("vec", bsonVector(v2)); + data2.put("name", new BsonString("two")); + upsertDoc(conn, tableName, "row1", new BsonDocument("data", data1)); + upsertDoc(conn, tableName, "row2", new BsonDocument("data", data2)); + upsertDoc(conn, tableName, "row3", + new BsonDocument("data", new BsonDocument("other", new BsonString("hello")))); + + // The region server extracts the vector. A row without the vector path gives SQL NULL. + String projSql = + "SELECT ID, BSON_VECTOR_VALUE(doc, 'data.vec', 3) FROM " + tableName + " ORDER BY ID"; + try (ResultSet rs = conn.createStatement().executeQuery("EXPLAIN " + projSql)) { + String plan = QueryUtil.getExplainPlan(rs); + assertTrue(plan, plan.contains("SERVER BSON PROJECTION 1")); + assertTrue(plan, plan.contains("BSON_VECTOR_VALUE(DOC, 'data.vec', 3)")); + } + try (ResultSet rs = conn.createStatement().executeQuery(projSql)) { + assertTrue(rs.next()); + assertEquals("row1", rs.getString(1)); + assertArrayEquals(v1, (float[]) rs.getObject(2), 1e-6f); + assertTrue(rs.next()); + assertEquals("row2", rs.getString(1)); + assertArrayEquals(v2, (float[]) rs.getObject(2), 1e-6f); + assertTrue(rs.next()); + assertEquals("row3", rs.getString(1)); + assertNull(rs.getObject(2)); + assertTrue(rs.wasNull()); + assertFalse(rs.next()); + } + + // The query lists a fixed-width vector value before a variable-width document value. The + // server serializes them in the other order, and the client must decode both values + // correctly. + String mixedSql = "SELECT ID, BSON_VECTOR_VALUE(doc, 'data.vec', 3)," + + " BSON_VALUE(doc, 'data.name', 'VARCHAR') FROM " + tableName + " WHERE ID = 'row2'"; + try (ResultSet rs = conn.createStatement().executeQuery(mixedSql)) { + assertTrue(rs.next()); + assertArrayEquals(v2, (float[]) rs.getObject(2), 1e-6f); + assertEquals("two", rs.getString(3)); + assertFalse(rs.next()); + } + + String distSql = "SELECT L2_DISTANCE(BSON_VECTOR_VALUE(doc, 'data.vec', 3), ARRAY[1.5, -2.5," + + " 0.003]) FROM " + tableName + " WHERE ID = 'row1'"; + try (ResultSet rs = conn.createStatement().executeQuery(distSql)) { + assertTrue(rs.next()); + assertEquals(0.0, rs.getDouble(1), 1e-6); + assertFalse(rs.next()); + } + + // If the query also projects the full document, the client extracts the vector from it + String fullDocSql = "SELECT doc, BSON_VECTOR_VALUE(doc, 'data.vec', 3) FROM " + tableName + + " WHERE ID = 'row1'"; + try (ResultSet rs = conn.createStatement().executeQuery("EXPLAIN " + fullDocSql)) { + String plan = QueryUtil.getExplainPlan(rs); + assertFalse(plan, plan.contains("SERVER BSON PROJECTION")); + } + try (ResultSet rs = conn.createStatement().executeQuery(fullDocSql)) { + assertTrue(rs.next()); + BsonDocument returnedDoc = (BsonDocument) rs.getObject(1); + assertArrayEquals(v1, + returnedDoc.getDocument("data").getBinary("vec").asVector().asFloat32Vector().getData(), + 0f); + assertArrayEquals(v1, (float[]) rs.getObject(2), 1e-6f); + assertFalse(rs.next()); + } + + // A dimension mismatch makes the evaluation on the region server fail + String badDim = "SELECT BSON_VECTOR_VALUE(doc, 'data.vec', 2) FROM " + tableName; + try (ResultSet rs = conn.createStatement().executeQuery(badDim)) { + while (rs.next()) { + rs.getObject(1); + } + fail("Expected evaluation exception on dimension mismatch"); + } catch (SQLException e) { + assertTrue(e.getMessage(), e.getMessage().contains("dimension mismatch")); + } + } + } +} diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIvfQueryIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIvfQueryIT.java index 2225bf850f..2506f615e5 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIvfQueryIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIvfQueryIT.java @@ -1800,7 +1800,11 @@ public class VectorIvfQueryIT extends ParallelStatsDisabledIT { exact.add(rs.getString(1) + "=" + rs.getString(2)); } } - assertEquals(exact, deferred); + List<String> expected = new ArrayList<>(); + expected.add("R0=n0"); + expected.add("R1=n1"); + assertEquals(expected, exact); + assertEquals(expected, deferred); } } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BsonVectorValueFunctionTest.java b/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BsonVectorValueFunctionTest.java new file mode 100644 index 0000000000..a461806b77 --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BsonVectorValueFunctionTest.java @@ -0,0 +1,332 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.expression.function; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.io.WritableUtils; +import org.apache.phoenix.expression.Determinism; +import org.apache.phoenix.expression.Expression; +import org.apache.phoenix.expression.ExpressionType; +import org.apache.phoenix.expression.LiteralExpression; +import org.apache.phoenix.parse.BsonVectorValueParseNode; +import org.apache.phoenix.schema.IllegalDataException; +import org.apache.phoenix.schema.types.PBson; +import org.apache.phoenix.schema.types.PDouble; +import org.apache.phoenix.schema.types.PInteger; +import org.apache.phoenix.schema.types.PJson; +import org.apache.phoenix.schema.types.PVarchar; +import org.apache.phoenix.schema.types.PVectorFloat; +import org.bson.BinaryVector; +import org.bson.BsonArray; +import org.bson.BsonBinary; +import org.bson.BsonBinarySubType; +import org.bson.BsonDocument; +import org.bson.BsonNull; +import org.bson.BsonString; +import org.bson.RawBsonDocument; +import org.junit.Test; + +public class BsonVectorValueFunctionTest { + + private static final float DELTA = 1e-6f; + + /** Returns a BSON binary value of subtype 9 (VECTOR) with FLOAT32 elements. */ + private static BsonBinary vector(float... values) { + return new BsonBinary(BinaryVector.floatVector(values)); + } + + /** Returns a subtype 9 binary value with the specified header bytes, to test validation. */ + private static BsonBinary rawVector(byte dtype, byte padding, float... values) { + ByteBuffer buf = ByteBuffer.allocate(2 + values.length * 4).order(ByteOrder.LITTLE_ENDIAN); + buf.put(dtype).put(padding); + for (float v : values) { + buf.putFloat(v); + } + return new BsonBinary(BsonBinarySubType.VECTOR, buf.array()); + } + + private static BsonVectorValueFunction func(BsonDocument doc, String path, int dim) + throws SQLException { + return new BsonVectorValueFunction( + Arrays.asList(LiteralExpression.newConstant(doc, PBson.INSTANCE), + LiteralExpression.newConstant(path, PVarchar.INSTANCE), + LiteralExpression.newConstant(dim, PInteger.INSTANCE))); + } + + private static float[] eval(BsonVectorValueFunction f) { + ImmutableBytesWritable ptr = new ImmutableBytesWritable(); + assertTrue("Function evaluation should succeed", f.evaluate(null, ptr)); + assertEquals(f.getDimension() * 4, ptr.getLength()); + float[] actual = (float[]) PVectorFloat.INSTANCE.toObject(ptr); + assertNotNull(actual); + return actual; + } + + private static void assertEvalFails(BsonVectorValueFunction f, String expectedFragment) { + try { + f.evaluate(null, new ImmutableBytesWritable()); + fail("Expected evaluation error containing '" + expectedFragment + "'"); + } catch (IllegalDataException e) { + assertTrue("Unexpected message: " + e.getMessage(), + e.getMessage().contains(expectedFragment)); + } + } + + @Test + public void testExtractsSpecEncodedVector() throws Exception { + // Verify extraction and big-endian transcoding against IEEE 754 float values. + float[] expected = new float[] { 1.5f, -2.5f, 3.0e-3f }; + BsonDocument doc = new BsonDocument("search", new BsonDocument("embedding", vector(expected))); + BsonVectorValueFunction f = func(doc, "search.embedding", 3); + + assertEquals(PVectorFloat.INSTANCE, f.getDataType()); + assertEquals(Integer.valueOf(3), f.getMaxLength()); + assertEquals(Determinism.ALWAYS, f.getDeterminism()); + assertArrayEquals(expected, eval(f), DELTA); + } + + @Test + public void testExtractsFromRawDocumentAsStoredInCell() throws Exception { + // The function must also read a RawBsonDocument, the form of a document in a stored cell + float[] expected = new float[] { 0.25f, 0.5f, 0.75f, 1.0f }; + BsonDocument doc = new BsonDocument("vec", vector(expected)); + byte[] bytes = PBson.INSTANCE.toBytes(doc); + RawBsonDocument raw = new RawBsonDocument(bytes); + assertArrayEquals(expected, eval(func(raw, "vec", 4)), DELTA); + } + + @Test + public void testExtractsFromArrayElementPath() throws Exception { + float[] first = new float[] { 1f, 2f }; + float[] second = new float[] { 3f, 4f }; + BsonDocument doc = new BsonDocument("chunks", new BsonArray(Arrays + .asList(new BsonDocument("emb", vector(first)), new BsonDocument("emb", vector(second))))); + assertArrayEquals(first, eval(func(doc, "chunks[0].emb", 2)), DELTA); + assertArrayEquals(second, eval(func(doc, "chunks[1].emb", 2)), DELTA); + } + + @Test + public void testMissingPathReturnsNull() throws Exception { + BsonDocument doc = + new BsonDocument("search", new BsonDocument("embedding", vector(1.0f, 2.0f))); + ImmutableBytesWritable ptr = new ImmutableBytesWritable(); + assertTrue("Missing path should evaluate to SQL NULL", + func(doc, "nonexistent.path", 2).evaluate(null, ptr)); + assertEquals(0, ptr.getLength()); + // A missing leaf under an existing parent document also gives SQL NULL + assertTrue(func(doc, "search.other", 2).evaluate(null, ptr)); + assertEquals(0, ptr.getLength()); + } + + @Test + public void testBsonNullAtPathReturnsNull() throws Exception { + BsonDocument doc = new BsonDocument("search", new BsonDocument("embedding", BsonNull.VALUE)); + ImmutableBytesWritable ptr = new ImmutableBytesWritable(); + assertTrue(func(doc, "search.embedding", 3).evaluate(null, ptr)); + assertEquals(0, ptr.getLength()); + } + + @Test + public void testNullDocumentReturnsNull() throws Exception { + ImmutableBytesWritable ptr = new ImmutableBytesWritable(); + assertTrue(func(null, "search.embedding", 3).evaluate(null, ptr)); + assertEquals(0, ptr.getLength()); + } + + @Test + public void testDimensionMismatchError() throws Exception { + BsonDocument doc = + new BsonDocument("search", new BsonDocument("embedding", vector(1.5f, 2.5f, 3.5f))); + assertEvalFails(func(doc, "search.embedding", 5), "dimension mismatch"); + assertEvalFails(func(doc, "search.embedding", 2), "dimension mismatch"); + } + + @Test + public void testWrongBinarySubtypeError() throws Exception { + // A binary value of subtype 0, not the vector subtype 9, is malformed + byte[] payload = vector(1.5f, 2.5f, 3.5f).getData(); + BsonDocument doc = new BsonDocument("search", + new BsonDocument("embedding", new BsonBinary(BsonBinarySubType.BINARY, payload))); + assertEvalFails(func(doc, "search.embedding", 3), "subtype"); + } + + @Test + public void testNonFloat32ElementTypeError() throws Exception { + // INT8 and PACKED_BIT vectors are malformed, because only FLOAT32 elements are supported + BsonBinary int8 = new BsonBinary(BinaryVector.int8Vector(new byte[] { 1, 2, 3, 4 })); + BsonDocument doc = new BsonDocument("v", int8); + assertEvalFails(func(doc, "v", 1), "FLOAT32"); + BsonBinary packed = + new BsonBinary(BinaryVector.packedBitVector(new byte[] { (byte) 0xF0 }, (byte) 4)); + assertEvalFails(func(new BsonDocument("v", packed), "v", 1), "FLOAT32"); + } + + @Test + public void testBadPaddingError() throws Exception { + BsonDocument doc = new BsonDocument("v", rawVector((byte) 0x27, (byte) 3, 1f, 2f)); + assertEvalFails(func(doc, "v", 2), "padding"); + } + + @Test + public void testNonFiniteElementError() throws Exception { + // A NaN or infinite element is malformed at any position in the payload + assertEvalFails(func(new BsonDocument("v", vector(1f, Float.NaN, 3f)), "v", 3), + "element 1 is NaN"); + assertEvalFails(func(new BsonDocument("v", vector(Float.NEGATIVE_INFINITY, 2f)), "v", 2), + "element 0 is -Infinity"); + assertEvalFails(func(new BsonDocument("v", vector(1f, 2f, Float.POSITIVE_INFINITY)), "v", 3), + "element 2 is Infinity"); + // Extreme finite values are valid + float[] extremes = { Float.MAX_VALUE, -Float.MAX_VALUE, Float.MIN_VALUE }; + assertArrayEquals(extremes, eval(func(new BsonDocument("v", vector(extremes)), "v", 3)), 0f); + } + + @Test + public void testHeaderlessPayloadError() throws Exception { + // A payload without the vector header is malformed + byte[] headerless = PVectorFloat.INSTANCE.toBytes(new float[] { 1f, 2f, 3f }); + BsonDocument doc = new BsonDocument("v", new BsonBinary(BsonBinarySubType.VECTOR, headerless)); + // Float bytes without the header fail the check of the element type byte + assertEvalFails(func(doc, "v", 3), "FLOAT32"); + BsonDocument empty = + new BsonDocument("v", new BsonBinary(BsonBinarySubType.VECTOR, new byte[1])); + assertEvalFails(func(empty, "v", 3), "truncated"); + } + + @Test + public void testNonBinaryValueAtPathError() throws Exception { + BsonDocument doc = + new BsonDocument("search", new BsonDocument("embedding", new BsonString("not_a_vector"))); + assertEvalFails(func(doc, "search.embedding", 3), "BSON Binary value"); + } + + @Test + public void testConstructorValidation() throws Exception { + Expression docExpr = LiteralExpression.newConstant(new BsonDocument(), PBson.INSTANCE); + Expression pathExpr = LiteralExpression.newConstant("path", PVarchar.INSTANCE); + for (int badDim : new int[] { -1, 0 }) { + try { + new BsonVectorValueFunction(Arrays.asList(docExpr, pathExpr, + LiteralExpression.newConstant(badDim, PInteger.INSTANCE))); + fail("Expected exception for dimension " + badDim); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("positive integer")); + } + } + try { + new BsonVectorValueFunction(Arrays.asList(docExpr, pathExpr, + LiteralExpression.newConstant(PVectorFloat.MAX_VECTOR_DIMENSION + 1, PInteger.INSTANCE))); + fail("Expected exception for dimension exceeding maximum"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("exceeds maximum")); + } + try { + new BsonVectorValueFunction(Arrays.asList(docExpr, pathExpr)); + fail("Expected exception for fewer than 3 arguments"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("requires 3 arguments")); + } + try { + new BsonVectorValueFunction( + Arrays.asList(docExpr, LiteralExpression.newConstant(null, PVarchar.INSTANCE), + LiteralExpression.newConstant(3, PInteger.INSTANCE))); + fail("Expected exception for null path"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("path")); + } + } + + @Test + public void testIntegrationWithDistanceFunction() throws Exception { + BsonDocument doc = + new BsonDocument("search", new BsonDocument("embedding", vector(3.0f, 0.0f))); + BsonVectorValueFunction f = func(doc, "search.embedding", 2); + // A distance function can use the extracted vector as an operand + Expression queryVec = + LiteralExpression.newConstant(new float[] { 0.0f, 4.0f }, PVectorFloat.INSTANCE); + L2DistanceFunction distFunc = new L2DistanceFunction(Arrays.asList(f, queryVec)); + ImmutableBytesWritable ptr = new ImmutableBytesWritable(); + assertTrue(distFunc.evaluate(null, ptr)); + assertEquals(5.0, ((Number) PDouble.INSTANCE.toObject(ptr)).doubleValue(), DELTA); + } + + @Test + public void testExpressionTypeSerializationRoundTrip() throws Exception { + float[] expected = new float[] { 1.5f, 2.5f, 3.5f }; + BsonDocument doc = new BsonDocument("search", new BsonDocument("embedding", vector(expected))); + BsonVectorValueFunction f = func(doc, "search.embedding", 3); + + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(baos); + WritableUtils.writeVInt(out, ExpressionType.valueOf(f).ordinal()); + f.write(out); + out.close(); + + DataInputStream in = new DataInputStream(new ByteArrayInputStream(baos.toByteArray())); + Expression deserialized = ExpressionType.values()[WritableUtils.readVInt(in)].newInstance(); + deserialized.readFields(in); + + assertTrue(deserialized instanceof BsonVectorValueFunction); + BsonVectorValueFunction df = (BsonVectorValueFunction) deserialized; + // The deserialized function must have the same path, dimension, and result as the original + assertEquals(3, df.getDimension()); + assertEquals("search.embedding", df.getPath()); + assertEquals(Integer.valueOf(3), df.getMaxLength()); + assertEquals(f, df); + assertArrayEquals(expected, eval(df), DELTA); + } + + @Test + public void testParseNodeCreation() throws Exception { + BsonVectorValueParseNode parseNode = + new BsonVectorValueParseNode("BSON_VECTOR_VALUE", Collections.emptyList(), null); + + Expression docBson = LiteralExpression.newConstant(new BsonDocument(), PBson.INSTANCE); + Expression docJson = LiteralExpression.newConstant("{}", PJson.INSTANCE); + Expression path = LiteralExpression.newConstant("embedding", PVarchar.INSTANCE); + Expression dim = LiteralExpression.newConstant(128, PInteger.INSTANCE); + + assertTrue( + parseNode.create(Arrays.asList(docBson, path, dim), null) instanceof BsonVectorValueFunction); + assertTrue( + parseNode.create(Arrays.asList(docJson, path, dim), null) instanceof BsonVectorValueFunction); + + Expression unsupportedDoc = LiteralExpression.newConstant(123, PInteger.INSTANCE); + try { + parseNode.create(Arrays.asList(unsupportedDoc, path, dim), null); + fail("Expected SQLException for unsupported data type"); + } catch (SQLException e) { + assertTrue(e.getMessage().contains("unsupported for BSON_VECTOR_VALUE")); + } + } +} 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 2a695ee2c5..29166cd53a 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 @@ -23,7 +23,9 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import java.io.IOException; import java.sql.Connection; import java.sql.Date; import java.sql.DriverManager; @@ -32,11 +34,13 @@ import java.sql.Time; import java.sql.Timestamp; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.HBaseConfiguration; @@ -52,8 +56,10 @@ 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.IndexRegionObserver; import org.apache.phoenix.hbase.index.ValueGetter; import org.apache.phoenix.hbase.index.covered.update.ColumnReference; +import org.apache.phoenix.hbase.index.table.HTableInterfaceReference; import org.apache.phoenix.hbase.index.util.GenericKeyValueBuilder; import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; import org.apache.phoenix.hbase.index.util.KeyValueBuilder; @@ -62,10 +68,12 @@ 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.IllegalDataException; 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.schema.types.PVectorFloat; import org.apache.phoenix.util.ByteUtil; import org.apache.phoenix.util.EnvironmentEdgeManager; import org.apache.phoenix.util.IndexUtil; @@ -74,8 +82,13 @@ 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.bson.BinaryVector; +import org.bson.BsonBinary; +import org.bson.BsonDocument; import org.junit.Test; +import org.apache.phoenix.thirdparty.com.google.common.collect.ArrayListMultimap; +import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap; import org.apache.phoenix.thirdparty.com.google.common.collect.Maps; public class IndexMaintainerTest extends BaseConnectionlessQueryTest { @@ -547,9 +560,100 @@ public class IndexMaintainerTest extends BaseConnectionlessQueryTest { 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)); + assertTrue( + im.isVectorUnchanged(im.getIndexedVector(near, false), im.getIndexedVector(same, false))); + assertFalse( + im.isVectorUnchanged(im.getIndexedVector(near, false), im.getIndexedVector(moved, false))); + assertFalse(im.isVectorUnchanged(im.getIndexedVector(noVector, false), + im.getIndexedVector(near, false))); + } + } + + private static Put bsonDataRow(PhoenixConnection pconn, String tableName, String id, + BsonDocument doc) throws Exception { + try (PreparedStatement ps = + pconn.prepareStatement("UPSERT INTO " + tableName + " VALUES (?, ?)")) { + ps.setString(1, id); + ps.setObject(2, doc); + ps.executeUpdate(); + } + Iterator<Pair<byte[], List<Mutation>>> iterator = pconn.getMutationState().toMutations(); + Put put = (Put) iterator.next().getSecond().get(0); + pconn.rollback(); + return put; + } + + @Test + public void testFunctionalVectorIndexMaintainer() 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 TABLE " + tableName + " (ID VARCHAR PRIMARY KEY, DOC BSON)"); + pconn.createStatement() + .execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + + " (BSON_VECTOR_VALUE(DOC, 'v', 2)) WITH (algorithm = 'IVF', metric = 'L2'," + + " lists = 2, sample_size = 10) ASYNC"); + PTable index = pconn.getTable(indexName); + PTable trained = new DelegateTable(index) { + @Override + public Long getVectorCentroidGeneration() { + return 3L; + } + }; + VectorCentroidCache.getInstance(HBaseConfiguration.create()).put(index.getName().getString(), + 3L, new CachedCentroids(Arrays.asList(new float[] { 0, 0 }, new float[] { 10, 10 }), + DistanceMetric.L2)); + IndexMaintainer im = IndexMaintainer.create(pconn.getTable(tableName), trained, pconn); + ColumnReference vectorColumn = im.getFunctionalVectorColumn(); + assertNotNull(vectorColumn); + + // The index row update must hold the computed vector in the PVectorFloat encoding + Put near = bsonDataRow(pconn, tableName, "a", + new BsonDocument("v", new BsonBinary(BinaryVector.floatVector(new float[] { 9, 8 })))); + 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()); + assertArrayEquals(PVectorFloat.INSTANCE.toBytes(new float[] { 9, 8 }), CellUtil + .cloneValue(indexPut.get(vectorColumn.getFamily(), vectorColumn.getQualifier()).get(0))); + IndexMaintainer fromProto = IndexMaintainer.fromProto(IndexMaintainer.toProto(im), + pconn.getTable(tableName).getRowKeySchema(), false); + assertEquals(vectorColumn, fromProto.getFunctionalVectorColumn()); + + // A malformed vector in the new row state makes the write fail. A malformed vector in the + // current row state gives no vector and no index row. + Put malformed = bsonDataRow(pconn, tableName, "b", + new BsonDocument("v", new BsonBinary(BinaryVector.int8Vector(new byte[] { 1, 2 })))); + assertNull(im.getIndexedVector(malformed, false)); + assertFalse(im.shouldPrepareIndexMutations(malformed, null)); + assertNull(im.buildRowKey(new IndexUtil.SimpleValueGetter(malformed), + new ImmutableBytesPtr(malformed.getRow()), null, null, ts)); + try { + im.shouldPrepareIndexMutations(malformed); + fail("A malformed vector being written must fail"); + } catch (IllegalDataException expected) { + } + try { + im.buildUpdateMutation(GenericKeyValueBuilder.INSTANCE, + new IndexUtil.SimpleValueGetter(malformed), new ImmutableBytesPtr(malformed.getRow()), ts, + null, null, false, null); + fail("A malformed vector being written must fail"); + } catch (IllegalDataException expected) { + } + + // A NaN element makes the vector malformed in the same way + Put nan = bsonDataRow(pconn, tableName, "c", new BsonDocument("v", + new BsonBinary(BinaryVector.floatVector(new float[] { 1, Float.NaN })))); + assertFalse(im.hasIndexRow(nan)); + try { + im.shouldPrepareIndexMutations(nan); + fail("A vector with a NaN element being written must fail"); + } catch (IllegalDataException expected) { + } } } @@ -690,4 +794,103 @@ public class IndexMaintainerTest extends BaseConnectionlessQueryTest { indexPut.getRow()); } } + + /** + * Returns a copy of a Put that counts the reads of one column. Each read is one evaluation of an + * expression on that column. + */ + private static Put countingReads(Put put, PColumn column, AtomicInteger reads) + throws IOException { + byte[] family = column.getFamilyName().getBytes(); + byte[] qualifier = column.getColumnQualifierBytes(); + return new Put(put) { + @Override + public List<Cell> get(byte[] f, byte[] q) { + if (Bytes.equals(family, f) && Bytes.equals(qualifier, q)) { + reads.incrementAndGet(); + } + return super.get(f, q); + } + }; + } + + private static List<Mutation> indexMutationsForRow(IndexMaintainer im, Put current, Put next, + long ts) throws IOException { + ListMultimap<HTableInterfaceReference, Mutation> updates = ArrayListMultimap.create(); + IndexRegionObserver.generateIndexMutationsForRow( + new ImmutableBytesPtr(next != null ? next.getRow() : current.getRow()), current, next, ts, + null, QueryConstants.UNVERIFIED_BYTES, Collections.singletonList(new Pair<>(im, + new HTableInterfaceReference(new ImmutableBytesPtr(im.getIndexTableName())))), + updates); + return new ArrayList<>(updates.values()); + } + + /** + * Verifies that a row write evaluates the indexed vector of each row state once, a document + * decode for a BSON vector, across every index mutation check of the write. + */ + @Test + public void testIndexedVectorEvaluatedOncePerRowState() 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 TABLE " + tableName + " (ID VARCHAR PRIMARY KEY, DOC BSON)"); + pconn.createStatement() + .execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + + " (BSON_VECTOR_VALUE(DOC, 'v', 2)) WITH (algorithm = 'IVF', metric = 'L2'," + + " lists = 2, sample_size = 10) ASYNC"); + PTable dataTable = pconn.getTable(tableName); + PTable trained = new DelegateTable(pconn.getTable(indexName)) { + @Override + public Long getVectorCentroidGeneration() { + return 3L; + } + }; + VectorCentroidCache.getInstance(HBaseConfiguration.create()).put(indexName, 3L, + new CachedCentroids(Arrays.asList(new float[] { 0, 0 }, new float[] { 10, 10 }), + DistanceMetric.L2)); + IndexMaintainer im = IndexMaintainer.create(dataTable, trained, pconn); + PColumn doc = dataTable.getColumnForColumnName("DOC"); + Put near = bsonDataRow(pconn, tableName, "a", + new BsonDocument("v", new BsonBinary(BinaryVector.floatVector(new float[] { 9, 8 })))); + Put far = bsonDataRow(pconn, tableName, "a", + new BsonDocument("v", new BsonBinary(BinaryVector.floatVector(new float[] { 1, 0 })))); + byte[] nearRowKey = ByteUtil.concat(PInteger.INSTANCE.toBytes(1), Bytes.toBytes("a")); + byte[] farRowKey = ByteUtil.concat(PInteger.INSTANCE.toBytes(0), Bytes.toBytes("a")); + long ts = EnvironmentEdgeManager.currentTimeMillis(); + AtomicInteger currentReads = new AtomicInteger(); + AtomicInteger nextReads = new AtomicInteger(); + + // An unchanged vector writes the index row again at the same row key + List<Mutation> mutations = indexMutationsForRow(im, countingReads(near, doc, currentReads), + countingReads(near, doc, nextReads), ts); + assertEquals(1, mutations.size()); + assertArrayEquals(nearRowKey, mutations.get(0).getRow()); + assertEquals(1, currentReads.get()); + assertEquals(1, nextReads.get()); + + // A changed vector moves the index row to its new centroid + currentReads.set(0); + nextReads.set(0); + mutations = indexMutationsForRow(im, countingReads(near, doc, currentReads), + countingReads(far, doc, nextReads), ts); + assertEquals(2, mutations.size()); + assertTrue(mutations.get(0) instanceof Put); + assertArrayEquals(farRowKey, mutations.get(0).getRow()); + assertTrue(mutations.get(1) instanceof Delete); + assertArrayEquals(nearRowKey, mutations.get(1).getRow()); + assertEquals(1, currentReads.get()); + assertEquals(1, nextReads.get()); + + // A data row delete also deletes the index row + currentReads.set(0); + mutations = indexMutationsForRow(im, countingReads(near, doc, currentReads), null, ts); + assertEquals(1, mutations.size()); + assertTrue(mutations.get(0) instanceof Delete); + assertArrayEquals(nearRowKey, mutations.get(0).getRow()); + assertEquals(1, currentReads.get()); + } + } } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/iterate/RegionScannerFactoryTest.java b/phoenix-core/src/test/java/org/apache/phoenix/iterate/RegionScannerFactoryTest.java new file mode 100644 index 0000000000..d9a1801c6a --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/iterate/RegionScannerFactoryTest.java @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.iterate; + +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.phoenix.query.QueryConstants; +import org.apache.phoenix.schema.PTable.QualifierEncodingScheme; +import org.apache.phoenix.schema.tuple.EncodedColumnQualiferCellsList; +import org.junit.Test; + +/** Unit tests for the position of the server parsed projection cell in a scanned row. */ +public class RegionScannerFactoryTest { + + private static final byte[] ROW = Bytes.toBytes("r"); + private static final byte[] CF = Bytes.toBytes("0"); + private static final QualifierEncodingScheme SCHEME = + QualifierEncodingScheme.FOUR_BYTE_QUALIFIERS; + + private static Cell column(int qualifier) { + return new KeyValue(ROW, CF, SCHEME.encode(qualifier), Bytes.toBytes(qualifier)); + } + + private static Cell arrayCell() { + return new KeyValue(ROW, QueryConstants.ARRAY_VALUE_COLUMN_FAMILY, + QueryConstants.ARRAY_VALUE_COLUMN_QUALIFIER, Bytes.toBytes("projected")); + } + + @Test + public void testArrayCellFollowsColumnsInOrdinaryList() { + List<Cell> cells = new ArrayList<>(); + cells.add(column(QueryConstants.ENCODED_EMPTY_COLUMN_NAME)); + cells.add(column(11)); + cells.add(arrayCell()); + assertEquals(2, RegionScannerFactory.getArrayCellPosition(cells)); + } + + @Test + public void testArrayCellFoundWithoutIndexedAccessToEncodedList() { + List<Cell> cells = + spy(new EncodedColumnQualiferCellsList(QueryConstants.ENCODED_CQ_COUNTER_INITIAL_VALUE, + QueryConstants.ENCODED_CQ_COUNTER_INITIAL_VALUE + 40, SCHEME)); + cells.add(column(QueryConstants.ENCODED_EMPTY_COLUMN_NAME)); + for (int q = QueryConstants.ENCODED_CQ_COUNTER_INITIAL_VALUE; q <= 50; q++) { + cells.add(column(q)); + } + // The reserved qualifier of the array cell sorts it after the empty column cell and before the + // other column cells + cells.add(arrayCell()); + assertEquals(1, RegionScannerFactory.getArrayCellPosition(cells)); + // The method must not use indexed access, because each indexed access scans the encoded list + // from the start + verify(cells, never()).get(anyInt()); + assertEquals(arrayCell().getValueLength(), cells.get(1).getValueLength()); + } +} diff --git a/phoenix-core/src/test/java/org/apache/phoenix/parse/VectorColumnParseTest.java b/phoenix-core/src/test/java/org/apache/phoenix/parse/VectorColumnParseTest.java index 56cdb5c9c4..22f35d4d1c 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/parse/VectorColumnParseTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/parse/VectorColumnParseTest.java @@ -350,4 +350,19 @@ public class VectorColumnParseTest { assertNull(indexStmt.getVectorLists()); assertNull(indexStmt.getVectorSampleSize()); } + + @Test + public void testParseBsonVectorValueFunctionInSelect() throws Exception { + String sql = "SELECT BSON_VECTOR_VALUE(doc, 'path', 128) FROM t"; + SQLParser parser = new SQLParser(sql); + BindableStatement stmt = parser.parseStatement(); + assertTrue("Expected SelectStatement", stmt instanceof SelectStatement); + SelectStatement selectStmt = (SelectStatement) stmt; + assertEquals(1, selectStmt.getSelect().size()); + ParseNode node = selectStmt.getSelect().get(0).getNode(); + assertTrue("Expected BsonVectorValueParseNode", node instanceof BsonVectorValueParseNode); + BsonVectorValueParseNode funcNode = (BsonVectorValueParseNode) node; + assertEquals("BSON_VECTOR_VALUE", funcNode.getName()); + assertEquals(3, funcNode.getChildren().size()); + } }
