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 68b53f93ec7fda24a2442ed002327bba748013d5 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 | 85 ++- .../BaseScannerRegionObserverConstants.java | 1 + .../coprocessorclient/MetaDataProtocol.java | 2 + .../apache/phoenix/expression/ExpressionType.java | 3 +- .../function/BsonVectorValueFunction.java | 221 ++++++++ .../org/apache/phoenix/index/IndexMaintainer.java | 156 +++++- .../org/apache/phoenix/iterate/ExplainTable.java | 3 +- .../apache/phoenix/optimize/OptimizerReasons.java | 2 + .../apache/phoenix/optimize/QueryOptimizer.java | 18 +- .../phoenix/parse/BsonVectorValueParseNode.java | 54 ++ .../parse/IndexExpressionParseNodeRewriter.java | 56 +- .../phoenix/query/ConnectionQueryServices.java | 3 +- .../phoenix/query/ConnectionQueryServicesImpl.java | 10 + .../src/main/protobuf/ServerCachingService.proto | 1 + .../phoenix/hbase/index/IndexRegionObserver.java | 5 +- .../iterate/NonAggregateRegionScannerFactory.java | 9 +- .../phoenix/iterate/RegionScannerFactory.java | 13 +- .../phoenix/end2end/BsonServerProjectionIT.java | 140 +++++ .../apache/phoenix/end2end/VectorBsonIndexIT.java | 575 +++++++++++++++++++++ .../apache/phoenix/end2end/VectorIvfQueryIT.java | 6 +- .../function/BsonVectorValueFunctionTest.java | 318 ++++++++++++ .../apache/phoenix/index/IndexMaintainerTest.java | 83 +++ .../phoenix/parse/VectorColumnParseTest.java | 15 + 23 files changed, 1706 insertions(+), 73 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 138f000e50..ff017a517f 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,13 +61,17 @@ 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; +import org.apache.phoenix.query.ConnectionQueryServices.Feature; import org.apache.phoenix.query.QueryConstants; import org.apache.phoenix.schema.AmbiguousColumnException; import org.apache.phoenix.schema.ArgumentTypeMismatchException; @@ -540,10 +547,12 @@ public class ProjectionCompiler { index++; } + // Preserve columns referenced by ORDER BY in the raw row rather than projecting server side + 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) { + // Evaluation layout order for 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); + // Match schema to the server's serialized projection field order + 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++) { @@ -861,14 +874,21 @@ public class ProjectionCompiler { return context.getSequenceManager().newSequenceReference(node); } + /** Evaluates whether BSON_VECTOR_VALUE projection is supported on target region servers. */ + private boolean isServerBsonVectorFunction(FunctionParseNode node) { + return BsonVectorValueFunction.NAME.equals(node.getName()) + && context.getConnection().getQueryServices().supportsFeature(Feature.VECTOR_INDEX); + } + @Override public Expression visitLeave(FunctionParseNode node, final List<Expression> children) throws SQLException { // this need not be done for group by clause with array or json. Hence, the below check if ( - !statement.isAggregate() && (ArrayIndexFunction.NAME.equals(node.getName()) - || isJsonFunction(node) || isBsonFunction(node)) + !statement.isAggregate() + && (ArrayIndexFunction.NAME.equals(node.getName()) || isJsonFunction(node) + || isBsonFunction(node) || isServerBsonVectorFunction(node)) && children.get(0) instanceof ProjectedColumnExpression ) { final List<KeyValueColumnExpression> indexKVs = Lists.newArrayList(); @@ -927,6 +947,43 @@ public class ProjectionCompiler { } } + /** Resolves column names referenced in the ORDER BY clause, expanding aliases and ordinals. */ + 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()); 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/coprocessorclient/MetaDataProtocol.java b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java index 3052f69b38..53ca5968aa 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java @@ -113,6 +113,8 @@ public abstract class MetaDataProtocol extends MetaDataService { public static final int MIN_VERSION_ALLOW_VBE_COLUMNS = VersionUtil.encodeVersion("5", "3", "0"); public static final int MIN_VERSION_TABLE_TTL_IN_SYSTEM_CATALOG = VersionUtil.encodeVersion("5", "3", "0"); + // Minimum server version supporting vector index maintenance and vector function evaluation + public static final int MIN_VECTOR_INDEX_VERSION = VersionUtil.encodeVersion("5", "4", "0"); // Version below which we should turn off essential column family. public static final int ESSENTIAL_FAMILY_VERSION_THRESHOLD = VersionUtil.encodeVersion("0", "94", "7"); 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..196031b8e1 --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/expression/function/BsonVectorValueFunction.java @@ -0,0 +1,221 @@ +/* + * 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"; + + /** Header size for BSON binary vectors (data type and padding bytes). */ + public static final int VECTOR_HEADER_SIZE = 2; + + private static final byte FLOAT32_DTYPE = BinaryVector.DataType.FLOAT32.getValue(); + + // Resolved constant arguments + private String path; + private int dimension; + + public BsonVectorValueFunction() { + // no-op + } + + 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)); + } + 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 eb21cbcfaf..d9c294edfc 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 @@ -92,6 +92,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; @@ -490,6 +491,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { private String vectorAlgorithm; private DistanceMetric distanceMetric; private Long centroidGeneration; + // Index column reference for expression derived vector values, or null if directly mapped. + private ColumnReference functionalVectorColumn; protected IndexMaintainer(RowKeySchema dataRowKeySchema, boolean isDataTableSalted) { this.dataRowKeySchema = dataRowKeySchema; @@ -520,6 +523,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 @@ -792,8 +800,19 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { return buildRowKey(valueGetter, rowKeyPtr, regionStartKey, regionEndKey, ts, null); } + /** + * Builds the index row key for a data row. Returns null for vector indexes if the vector + * evaluates to null or cannot be read. + */ public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr, byte[] regionStartKey, byte[] regionEndKey, long ts, byte[] encodedRegionName) { + return buildRowKey(valueGetter, rowKeyPtr, regionStartKey, regionEndKey, ts, encodedRegionName, + null); + } + + 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"); } @@ -875,16 +894,13 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { dataSortOrder = expression.getSortOrder(); isNullable = expression.isNullable(); if (i == 0 && isVectorIndex()) { - // 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; @@ -1166,7 +1182,18 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { */ public boolean shouldPrepareIndexMutations(Put dataRowState) { - if (isVectorIndex() && getIndexedVector(dataRowState) == null) { + return shouldPrepareIndexMutations(dataRowState, true); + } + + /** + * Checks whether the current row state corresponds to an existing index entry for deletion. + */ + public boolean hasIndexRow(Put currentDataRowState) { + return shouldPrepareIndexMutations(currentDataRowState, false); + } + + private boolean shouldPrepareIndexMutations(Put dataRowState, boolean strict) { + if (isVectorIndex() && getIndexedVector(dataRowState, strict) == null) { // Null vectors omit index mutations return false; } @@ -1377,15 +1404,24 @@ 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 { + // Strict evaluation rejects malformed vector payloads during row writes + ImmutableBytesWritable indexedVector = null; + if (isVectorIndex()) { + indexedVector = getIndexedVector(valueGetter, ts, true); + if (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, @@ -1395,6 +1431,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 @@ -1434,6 +1484,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { } familyToColListMap.get(cf).add(Pair.newPair(indexColRef, ref)); } + if (computedValue != null) { + // Expression-derived columns lack corresponding source data table columns + 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()) { @@ -1452,7 +1507,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() { @@ -1602,6 +1661,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; } @@ -1626,6 +1693,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()) + ) { + // Exclude computed vector column from covered column tally + 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 @@ -2140,7 +2214,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; @@ -2304,7 +2384,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(); } @@ -2705,6 +2792,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { return centroidGeneration; } + /** Returns the column reference for computed vector expressions, or null. */ + public ColumnReference getFunctionalVectorColumn() { + return functionalVectorColumn; + } + /** * Pre-warms the centroid cache for this vector index generation prior to row lock acquisition. If * not cached, loads centroids via {@code conn} or internal server connection. @@ -2725,17 +2817,35 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { } /** Extracts the indexed vector value from the row state, or null if absent. */ - private ImmutableBytesWritable getIndexedVector(Put dataRowState) { + private ImmutableBytesWritable getIndexedVector(Put dataRowState, boolean strict) { if (dataRowState == null) { return null; } + return getIndexedVector(new IndexUtil.SimpleValueGetter(dataRowState), + HConstants.LATEST_TIMESTAMP, strict); + } + + /** + * Extracts the indexed vector value. Evaluates strictly during row mutations to reject malformed + * vectors, and leniently when reading existing row state to permit deletes and updates. + */ + 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; @@ -2749,8 +2859,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { if (!isVectorIndex()) { return false; } - ImmutableBytesWritable current = getIndexedVector(currentDataRowState); - ImmutableBytesWritable next = getIndexedVector(nextDataRowState); + ImmutableBytesWritable current = getIndexedVector(currentDataRowState, false); + ImmutableBytesWritable next = getIndexedVector(nextDataRowState, false); 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 31e6896420..5ee0235613 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 @@ -54,6 +54,8 @@ public final class OptimizerReasons { public static final String REASON_NOT_A_VECTOR_SEARCH = "not a nearest-neighbor search"; public static final String REASON_VECTOR_METRIC_MISMATCH = "distance metric does not match index"; 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 5361d2d180..9a8f2b147a 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 @@ -599,8 +599,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()) { @@ -820,11 +823,16 @@ public class QueryOptimizer { * Checks whether the rewritten index query orders results using the index's indexed vector * column. */ - 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()) { + // Query ranks by an expression not covered by the target vector index + 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 d7022febb5..f9735d5822 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,44 @@ public class IndexExpressionParseNodeRewriter extends ParseNodeRewriter { // Centroid IDs are computed from indexed vectors and do not map to query expressions 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()) { + // Functional vector indexes store the computed vector in a non-PK data column + PColumn vectorColumn = VectorIndexTrainer.getIndexedVectorColumn(index); + if (!(SQLParser.parseCondition(vectorColumn.getExpressionStr()) instanceof ColumnParseNode)) { + addMapping(vectorColumn, alias, expressionCompiler, rewriter); } } } + /** Registers expression-to-column rewrite mappings for the specified index 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 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() }); + } + } + @Override protected ParseNode leaveCompoundNode(CompoundParseNode node, List<ParseNode> children, CompoundNodeFactory factory) { diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java index 8808d67d76..d52ef81e6c 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java @@ -234,7 +234,8 @@ public interface ConnectionQueryServices extends QueryServices, MetaDataMutated public enum Feature { LOCAL_INDEX, - RENEW_LEASE + RENEW_LEASE, + VECTOR_INDEX }; public boolean supportsFeature(Feature feature); diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java index bd24c7b392..bd22f0625c 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java @@ -377,6 +377,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices // Lowest HBase version on the cluster. private int lowestClusterHBaseVersion = Integer.MAX_VALUE; + private int lowestClusterPhoenixVersion = Integer.MAX_VALUE; private boolean hasIndexWALCodec = true; @GuardedBy("connectionCountLock") @@ -460,6 +461,11 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices int hbaseVersion = services.getLowestClusterHBaseVersion(); return hbaseVersion >= MetaDataProtocol.MIN_RENEW_LEASE_VERSION; } + }, Feature.VECTOR_INDEX, new FeatureSupported() { + @Override + public boolean isSupported(ConnectionQueryServices services) { + return lowestClusterPhoenixVersion >= MetaDataProtocol.MIN_VECTOR_INDEX_VERSION; + } }); private QueryLoggerDisruptor queryDisruptor; @@ -2167,6 +2173,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices throws SQLException, AccessDeniedException { StringBuilder errorMessage = new StringBuilder(); int minHBaseVersion = Integer.MAX_VALUE; + int minPhoenixVersion = Integer.MAX_VALUE; boolean isTableNamespaceMappingEnabled = false; long systemCatalogTimestamp = Long.MAX_VALUE; long startTime = 0L; @@ -2235,6 +2242,8 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices if (minHBaseVersion > MetaDataUtil.decodeHBaseVersion(serverJarVersion)) { minHBaseVersion = MetaDataUtil.decodeHBaseVersion(serverJarVersion); } + minPhoenixVersion = + Math.min(minPhoenixVersion, MetaDataUtil.decodePhoenixVersion(serverJarVersion)); // In case this is the first time connecting to this cluster, the system catalog table does // not have an // entry for itself yet, so we cannot get the timestamp and this will not be returned from @@ -2272,6 +2281,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices .build().buildException(); } lowestClusterHBaseVersion = minHBaseVersion; + lowestClusterPhoenixVersion = minPhoenixVersion; } finally { if (ht != null) { try { 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 19f2fdaf14..cc07912a39 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 @@ -1437,10 +1437,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver { indexUpdates.put(hTableInterfaceReference, del); } } - } else if ( - currentDataRowState != null - && indexMaintainer.shouldPrepareIndexMutations(currentDataRowState) - ) { + } else if (currentDataRowState != null && indexMaintainer.hasIndexRow(currentDataRowState)) { if (indexMaintainer.isCDCIndex()) { // CDC Index needs two a delete marker for referencing the data table // delete mutation with the right index row key, that is, the index row key 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..0b48b6030d 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 @@ -473,8 +473,19 @@ public abstract class RegionScannerFactory { }; } - // PHOENIX-4791 Share position of array element cell + // PHOENIX-4791 Share position of array element cell. The cell is appended last to an ordinary + // list, but an encoded qualifier list keeps it in its reserved qualifier range, ahead of the + // column cells, so look it up by its column. public static int getArrayCellPosition(List<Cell> result) { + for (int i = result.size() - 1; i >= 0; i--) { + Cell cell = result.get(i); + if ( + CellUtil.matchingColumn(cell, QueryConstants.ARRAY_VALUE_COLUMN_FAMILY, + QueryConstants.ARRAY_VALUE_COLUMN_QUALIFIER) + ) { + return 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..71e3a5e2c7 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BsonServerProjectionIT.java @@ -0,0 +1,140 @@ +/* + * 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 with top-N ordering and multi-function + * evaluation. + */ +@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 combined with TOP-N row ordering and 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 server side BSON projection with non-encoded column storage scheme. */ + @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 ORDER BY expressions referencing the projected BSON document column. */ + @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 correct decoding order when multiple distinct BSON functions are projected. */ + @Test + public void testMixedFunctionProjection() throws Exception { + try (Connection conn = DriverManager.getConnection(getUrl())) { + String table = createTable(conn); + List<String> rows = rows(conn, + "SELECT ID, BSON_VALUE(DOC, 'rank', 'INTEGER'), BSON_VALUE(DOC, 'name', 'VARCHAR')," + + " BSON_VALUE(DOC, 'tag.k', 'VARCHAR') FROM " + table + " WHERE ID = 'id2'"); + assertEquals(1, rows.size()); + assertEquals("id2,8,name2,tag2", rows.get(0)); + assertFalse(rows.get(0).contains("null")); + } + } +} 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..f454f7cae5 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorBsonIndexIT.java @@ -0,0 +1,575 @@ +/* + * 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 extraction via BSON_VECTOR_VALUE, functional vector indexing, + * query optimization, 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)); + } + + /** Generates a test vector with the specified leading component. */ + 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()); + } + } + + /** Test centroids partitioned along 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; + } + + /** Maps indexed row IDs to their assigned centroid IDs. */ + 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); + + // Functional vector indexes store the computed vector directly in the index row + 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()); + + // Seed test rows spanning centroids as well as null and missing vector cases + 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())); + } + + // Verify persisted vector encoding matches big endian PVectorFloat layout + 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())); + } + + // Non-vector field mutation preserves existing 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")); + + // Vector value mutation relocates index entry to the newly assigned 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")); + + // Vector deletion removes the corresponding index entry + 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")); + + // Malformed vector payload rejects write operation + 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 preexisting unindexed rows with malformed vectors remain deletable and updateable + * without failing maintenance validation. + */ + @Test + public void testMalformedCurrentVectorDoesNotBlockDeleteOrCorrection() throws Exception { + String tableName = "T_BSON_BAD_" + generateUniqueName(); + String indexName = "IDX_BSON_BAD_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + // Untrained index on empty table omits maintenance operations + createIndexedTable(conn, tableName, indexName); + BsonDocument bad = + new BsonDocument("embedding", new BsonBinary(BinaryVector.int8Vector(new byte[DIM]))); + upsertDoc(conn, tableName, "fix", bad); + upsertDoc(conn, tableName, "del", bad); + 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(); + try { + upsertDoc(conn, tableName, "keep", bad); + 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()); + + // Populate boundary vectors across centroid thresholds + 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)); + } + + // Validate commutative distance operand parsing and plan selection + 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)); + } + } + + /** + * Ensures covered vector columns with differing dimensions do not collide with indexed vectors. + */ + @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 index rejection when query path, dimension, or target column mismatches. */ + @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 with server-side projections. + */ + @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")))); + + // Verify pushdown of vector extraction to region server + 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()); + } + + // Projection ordering: fixed-width vector column preceding variable-width document fields + 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()); + } + + // Full document projection delegates vector extraction to client side evaluation + 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()); + } + + // Region server evaluation failure on vector dimension mismatch + 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 6e4d6fcf8b..3d492ac05d 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 @@ -1665,7 +1665,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..8a6c22fb77 --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/expression/function/BsonVectorValueFunctionTest.java @@ -0,0 +1,318 @@ +/* + * 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; + + /** Constructs a BSON binary vector payload with subtype 9 and FLOAT32 encoding. */ + private static BsonBinary vector(float... values) { + return new BsonBinary(BinaryVector.floatVector(values)); + } + + /** Constructs a custom subtype 9 binary payload for validation testing. */ + 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 { + // Verify extraction from lazily-decoded RawBsonDocument instances. + 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()); + // Missing leaf path within existing parent document. + 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 { + // Binary subtype 0 instead of vector subtype 9 + 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 { + // Reject non-FLOAT32 vector element types. + 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 testHeaderlessPayloadError() throws Exception { + // Payloads lacking the required vector header reject evaluation + byte[] headerless = PVectorFloat.INSTANCE.toBytes(new float[] { 1f, 2f, 3f }); + BsonDocument doc = new BsonDocument("v", new BsonBinary(BsonBinarySubType.VECTOR, headerless)); + // Un-prefixed float bytes fail element type header validation. + 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); + // Evaluate distance calculation against extracted vector. + 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; + // Verify reconstructed expression and metadata state + 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 fbceef8ce5..5305c2f986 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,6 +23,7 @@ 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.Date; @@ -61,10 +62,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; @@ -72,6 +75,9 @@ import org.apache.phoenix.util.PhoenixRuntime; import org.apache.phoenix.util.PropertiesUtil; import org.apache.phoenix.util.SchemaUtil; import org.apache.phoenix.util.TestUtil; +import org.bson.BinaryVector; +import org.bson.BsonBinary; +import org.bson.BsonDocument; import org.junit.Test; import org.apache.phoenix.thirdparty.com.google.common.collect.Maps; @@ -551,6 +557,83 @@ public class IndexMaintainerTest extends BaseConnectionlessQueryTest { } } + 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); + + // Verify index mutation contains computed vector encoded as PVectorFloat + 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()); + + // Malformed vectors reject writes during update, but omit index entries for current + // state evaluation + Put malformed = bsonDataRow(pconn, tableName, "b", + new BsonDocument("v", new BsonBinary(BinaryVector.int8Vector(new byte[] { 1, 2 })))); + assertFalse(im.hasIndexRow(malformed)); + 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) { + } + } + } + @Test public void testNonVectorIndexMaintainerHasNoVectorFields() throws Exception { String tableName = "T_" + generateUniqueName(); 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 f8c96ba149..aa80f4ae29 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()); + } }
