Copilot commented on code in PR #2624: URL: https://github.com/apache/phoenix/pull/2624#discussion_r4209166764
########## phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexRebuilder.java: ########## @@ -0,0 +1,444 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.index.vector; + +import static org.apache.phoenix.query.QueryConstants.TRUE; + +import java.io.IOException; +import java.sql.SQLException; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.ResultScanner; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.filter.KeyOnlyFilter; +import org.apache.phoenix.cache.VectorCentroidCache; +import org.apache.phoenix.cache.VectorCentroidCache.CachedCentroids; +import org.apache.phoenix.compile.MutationPlan; +import org.apache.phoenix.compile.ServerBuildIndexCompiler; +import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants; +import org.apache.phoenix.exception.SQLExceptionCode; +import org.apache.phoenix.exception.SQLExceptionInfo; +import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.optimize.DistanceMetric; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTable.TaskType; +import org.apache.phoenix.schema.TableNotFoundException; +import org.apache.phoenix.util.ClientUtil; +import org.apache.phoenix.util.EnvironmentEdgeManager; +import org.apache.phoenix.util.ReadOnlyProps; +import org.apache.phoenix.util.SchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Generational rebuild orchestration and drift reconciliation for IVF vector indexes. + * <p> + * Rebuilding trains a new centroid generation, marks the index in migrating state via + * {@code SYSTEM.CATALOG}, and performs an online migration across centroid generations. Concurrent + * writes target the building generation while purging prior index rows under either generation. + * Online migration executes via a server side index build followed by catch up scan repair of + * concurrent mutations, atomic generation promotion, and retirement of prior generations. + * <p> + * Distributed mutual exclusion is maintained using atomic lease rows in + * {@code SYSTEM.VECTOR_CENTROID}. Interrupted rebuilds preserve migration state to enable + * idempotent resumption during periodic reconciliation tasks. + */ +public final class VectorIndexRebuilder { + + private static final Logger LOGGER = LoggerFactory.getLogger(VectorIndexRebuilder.class); + + /** Trigger reason identifier for operator initiated index rebuilds. */ + public static final String MANUAL_REASON = "MANUAL"; + /** Maximum lease duration for rebuild locks before allowing preemption. */ + public static final long CLAIM_EXPIRY_MS = 24L * 60 * 60 * 1000; Review Comment: The rebuild claim expires after 24 hours but neither rebuild nor fsck repair renews it while work is running. A large backfill or repair can exceed this duration, allowing another process to preempt the still-active operation and concurrently mutate generation metadata/postings. Add lease renewal with ownership checks, and abort safely if renewal is lost. ########## phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/VectorIndexTrainer.java: ########## @@ -0,0 +1,210 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.index.vector; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.Random; +import org.apache.phoenix.cache.VectorCentroidCache; +import org.apache.phoenix.cache.VectorCentroidCache.CachedCentroids; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.optimize.DistanceMetric; +import org.apache.phoenix.schema.PColumn; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.util.SchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Coordinates IVF index centroid training, persistence, and catalog registration across index + * creation and rebuild operations. + * <p> + * Samples training vectors from the base table expression using server side Bernoulli filtering + * combined with client side reservoir sampling, then executes in-memory k-means clustering. + */ +public final class VectorIndexTrainer { + + private static final Logger LOGGER = LoggerFactory.getLogger(VectorIndexTrainer.class); + /** Oversampling margin to ensure sufficient sample volume for the reservoir. */ + private static final double OVERSAMPLE = 1.2; + /** Initial creation trigger reason for newly trained vector indexes. */ + public static final String CREATE_INDEX_REASON = "CREATE_INDEX"; + + private VectorIndexTrainer() { + } + + /** + * Identifies the primary vector column definition within the index schema. + */ + public static PColumn getIndexedVectorColumn(PTable index) { + for (PColumn column : index.getColumns()) { + if ( + column.getFamilyName() != null && column.getDataType() != null + && column.getDataType().isVectorType() + ) { + return column; + } + } + throw new IllegalStateException("Vector index " + index.getName() + " has no vector column"); + } + + /** Samples base table vectors and trains centroids according to the index configuration. */ + public static KMeansResult train(PhoenixConnection conn, PTable dataTable, PTable index) + throws SQLException { + return train(conn, dataTable, index, index.getVectorIvfLists()); + } + + /** + * Samples base table vectors and trains the requested number of centroids using the index + * configuration. + * @return trained k-means model, or null if sample volume is insufficient for the requested + * clusters + */ + public static KMeansResult train(PhoenixConnection conn, PTable dataTable, PTable index, + int lists) throws SQLException { + int sampleSize = index.getVectorIvfSampleSize(); + DistanceMetric metric = DistanceMetric.fromString(index.getVectorDistanceMetric()); + PColumn vectorColumn = getIndexedVectorColumn(index); + String expr = vectorColumn.getExpressionStr(); + String from = SchemaUtil.getEscapedFullTableName(dataTable.getName().getString()); + String notNull = " WHERE " + expr + " IS NOT NULL"; + + long count; + try (Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM " + from + notNull)) { + count = rs.next() ? rs.getLong(1) : 0; + } + if (count < lists) { + LOGGER.info("Deferring centroid training for vector index {}: {} non-null vectors, fewer than" + + " lists={}", index.getName(), count, lists); + return null; + } + double p = Math.min(1.0, OVERSAMPLE * sampleSize / count); + String sql = "SELECT " + expr + " FROM " + from + notNull; + List<float[]> sample = sample(conn, p < 1.0 ? sql + " AND RAND() < " + p : sql, sampleSize); + if (sample.size() < lists) { + // Fallback to full scan when Bernoulli sampling yields fewer rows than requested lists + sample = sample(conn, sql, sampleSize); + } + KMeansConfig config = KMeansConfig.newBuilder().distanceMetric(metric).build(); + LOGGER.info("Training vector index {} with {} sampled vectors, lists={}, metric={}", + index.getName(), sample.size(), lists, metric); + return KMeansTrainer.train(sample, lists, config); + } + + /** + * Trains a new centroid generation, persists centroid vectors, updates catalog metadata, and + * installs the model into the local centroid cache. + */ + public static Long trainAndRecord(PhoenixConnection conn, PTable dataTable, PTable index) + throws SQLException { + KMeansResult result = train(conn, dataTable, index); + if (result == null) { + return null; + } + String indexName = index.getName().getString(); + long generation = CentroidManager.nextGeneration(index.getVectorCentroidGeneration()); + CentroidManager.persistCentroids(conn, indexName, generation, result.getCentroids()); + CentroidManager.persistGenerationSummary(conn, indexName, generation, + new GenerationSummary(GenerationSummary.ACTIVE, CREATE_INDEX_REASON, result.getRequestedK(), + result.getSkewMetrics(), null, null)); + CentroidManager.setGenerationAndLists(conn, index, generation, result.getEffectiveK()); + VectorCentroidCache.getInstance(conn.getQueryServices().getConfiguration()).put(indexName, + generation, new CachedCentroids(result.getCentroids(), result.getDistanceMetric())); + return generation; + } + + private static List<float[]> sample(PhoenixConnection conn, String sql, int sampleSize) + throws SQLException { + try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(sql)) { + return KMeansTrainer.reservoirSample(new VectorIterator(rs), sampleSize, new Random()); + } catch (VectorIterator.ReadException e) { + throw e.getCause(); + } + } + + /** + * Converts a JDBC vector value ({@code float[]} or {@code double[]}) to a float array. + * @param value vector object read via JDBC + * @return float array representation, or null if value is null + */ + public static float[] toFloats(Object value) { + if (value instanceof float[]) { + return (float[]) value; + } + if (value instanceof double[]) { + double[] d = (double[]) value; + float[] f = new float[d.length]; + for (int i = 0; i < d.length; i++) { + f[i] = (float) d[i]; + } Review Comment: This downcasts every `VECTOR(DOUBLE, ...)` training sample to float, even though double vectors are accepted for IVF indexes. Values outside the float range become infinities (and can make distance calculations NaN), while distinct double values can collapse to the same float; the cache also performs the same conversion during writes and probes. Preserve double precision end-to-end or explicitly reject double vector indexes. ########## phoenix-core-client/src/main/java21/org/apache/phoenix/expression/function/PanamaDistanceKernel.java: ########## @@ -0,0 +1,136 @@ +/* + * 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 jdk.incubator.vector.ByteVector; +import jdk.incubator.vector.FloatVector; +import jdk.incubator.vector.VectorOperators; +import jdk.incubator.vector.VectorSpecies; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.phoenix.schema.types.PVectorFloat; + +/** + * SIMD distance kernels leveraging the Panama Vector API over packed little-endian float vectors. + * <p> + * Operands are loaded directly from their {@code byte[]} encodings, at any offset, as byte vectors + * reinterpreted to float lanes. The Vector API defines that reinterpretation as little-endian on + * every platform, which matches the {@link PVectorFloat} encoding, so no decoding pass or copy + * precedes the arithmetic. + * <p> + * Accumulates intermediate lane sums in single precision across fixed size blocks of + * {@link #BLOCK_STRIDES} vector strides before reducing to double precision. Block level reduction + * bounds floating point accumulation drift relative to {@link ScalarDistanceKernel} while enabling + * periodic upper bound pruning without per-stride reduction overhead. + */ +final class PanamaDistanceKernel { + + private static final VectorSpecies<Float> SPECIES = FloatVector.SPECIES_PREFERRED; + private static final VectorSpecies<Byte> BYTE_SPECIES = + VectorSpecies.of(byte.class, SPECIES.vectorShape()); + static final int BLOCK_STRIDES = 8; + + private PanamaDistanceKernel() { + } + + private static FloatVector load(byte[] buf, int off, int i) { + return ByteVector.fromArray(BYTE_SPECIES, buf, off + i * Bytes.SIZEOF_FLOAT) + .reinterpretAsFloats(); + } + + static double l2DistanceSquared(byte[] a, int aOff, byte[] b, int bOff, int dim, double bound) { + int step = SPECIES.length(); + int upper = SPECIES.loopBound(dim); + int blockLen = step * BLOCK_STRIDES; + double total = 0.0; + int i = 0; + while (i < upper) { + int blockEnd = Math.min(upper, i + blockLen); + FloatVector acc = FloatVector.zero(SPECIES); + for (; i < blockEnd; i += step) { + FloatVector diff = load(a, aOff, i).sub(load(b, bOff, i)); + acc = diff.fma(diff, acc); + } + total += acc.reduceLanes(VectorOperators.ADD); Review Comment: The SIMD lanes multiply and accumulate in `float`, unlike the scalar reference which promotes each operand to `double`. For finite inputs such as `Float.MAX_VALUE`, `diff * diff` overflows to infinity here while the scalar result remains finite (~1e77), so Java 21 can return different distances and rankings from other runtimes. The dot-product and cosine vector loops have the same issue; accumulate products with double precision (or fall back before float overflow). ########## phoenix-core-client/src/main/java/org/apache/phoenix/index/vector/KMeansTrainer.java: ########## @@ -0,0 +1,342 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.index.vector; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import java.util.Random; +import org.apache.phoenix.expression.function.VectorDistanceUtil; +import org.apache.phoenix.optimize.DistanceMetric; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * In-memory k-means trainer for IVF index centroids. + * <p> + * Provides k-means++ initialization, metric aware assignment and centroid updates (standard Lloyd + * updates for L2 and inner product; spherical k-means for cosine), empty cluster reseeding, + * relative distortion convergence, and post-training cluster splitting to balance posting lists. + * Assignment distance calculation is centralized in {@link #assignmentDistance} to guarantee + * consistent routing across training, index maintenance, and query probing. + */ +public final class KMeansTrainer { + + private static final Logger LOGGER = LoggerFactory.getLogger(KMeansTrainer.class); + + private KMeansTrainer() { + } + + /** + * Computes vector assignment distance according to the specified distance metric. Cosine distance + * is used for COSINE indexes; squared Euclidean distance is used for L2 and INNER_PRODUCT + * indexes. + */ + public static double assignmentDistance(DistanceMetric metric, float[] a, float[] b) { + if (metric == DistanceMetric.COSINE) { + return VectorDistanceUtil.scalarCosineDistance(a, b, a.length); + } + return VectorDistanceUtil.scalarL2DistanceSquared(a, b, a.length); Review Comment: `INNER_PRODUCT` is routed through squared L2 distance, so centroid training and probing optimize a different metric from the query. For example, a high-magnitude vector with the best inner product can be placed in a centroid that L2 probing does not visit, severely reducing recall. Implement a metric-consistent assignment/seeding strategy for inner-product indexes rather than silently treating them as L2. ########## phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PVectorDataType.java: ########## @@ -0,0 +1,351 @@ +/* + * 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.schema.types; + +import java.lang.reflect.Array; +import java.sql.SQLException; +import java.text.Format; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.phoenix.schema.ConstraintViolationException; +import org.apache.phoenix.schema.SortOrder; +import org.apache.phoenix.util.ByteUtil; + +/** + * Base class for fixed-dimension vector types {@code VECTOR(FLOAT, N)} and + * {@code VECTOR(DOUBLE, N)}. + * <p> + * Vectors are serialized as dense, contiguous IEEE 754 big-endian values without header metadata or + * null masks. Vector dimension is stored in column max length. Vectors do not support primary key Review Comment: This documents the storage encoding as big-endian, but both vector implementations reverse bytes and explicitly store little-endian values. Correcting this is important because external producers and BSON interoperability depend on the byte order. This issue also appears on line 68 of the same file. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
