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 25acdb5ad8c69da805bb8194f528700f6876ec7f Author: Andrew Purtell <[email protected]> AuthorDate: Tue Oct 6 00:07:16 2026 -0700 PHOENIX-7998 Phase 11: Single-cell storage support for vector indexes Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> --- .../org/apache/phoenix/index/IndexMaintainer.java | 4 +- .../apache/phoenix/end2end/VectorIndexWriteIT.java | 183 ++++++++++++++++++--- .../apache/phoenix/end2end/VectorIvfQueryIT.java | 143 ++++++++++++++++ .../apache/phoenix/index/IndexMaintainerTest.java | 39 +++++ 4 files changed, 344 insertions(+), 25 deletions(-) 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 b1b5e238c5..4ab1029049 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 @@ -1576,8 +1576,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> { public PDataType getDataType() { return null; } - }, dataColRef.getFamily(), dataColRef.getQualifier(), destEncodingScheme, - destImmutableStorageScheme); + }, dataColRef.getFamily(), dataColRef.getQualifier(), srcEncodingScheme, + srcImmutableStorageScheme); ImmutableBytesPtr ptr = new ImmutableBytesPtr(); expression.evaluate(new ValueGetterTuple(valueGetter, ts), ptr); value = ptr.copyBytesIfNecessary(); diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexWriteIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexWriteIT.java index b8a9b4baf7..5d7cc213a8 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexWriteIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VectorIndexWriteIT.java @@ -31,12 +31,16 @@ 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.Assume.assumeFalse; 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.Arrays; +import java.util.Collection; import java.util.List; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.client.Get; @@ -47,11 +51,15 @@ import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.util.Bytes; import org.apache.phoenix.end2end.index.IndexTestUtil; +import org.apache.phoenix.expression.SingleCellColumnExpression; +import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.query.QueryConstants; import org.apache.phoenix.schema.PColumn; import org.apache.phoenix.schema.PIndexState; import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTable.ImmutableStorageScheme; +import org.apache.phoenix.schema.tuple.ResultTuple; import org.apache.phoenix.schema.types.PInteger; import org.apache.phoenix.schema.types.PVectorDouble; import org.apache.phoenix.schema.types.PVectorFloat; @@ -61,6 +69,8 @@ import org.apache.phoenix.util.IndexUtil; import org.apache.phoenix.util.SchemaUtil; import org.junit.Test; import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; /** * Integration tests for vector index maintenance on the write path. The tests cover the centroid @@ -69,15 +79,93 @@ import org.junit.experimental.categories.Category; * for each immutable storage scheme. */ @Category(ParallelStatsDisabledTest.class) +@RunWith(Parameterized.class) public class VectorIndexWriteIT extends ParallelStatsDisabledIT { + private final ImmutableStorageScheme storageScheme; + + public VectorIndexWriteIT(ImmutableStorageScheme storageScheme) { + this.storageScheme = storageScheme; + } + + @Parameterized.Parameters(name = "VectorIndexWriteIT_storageScheme={0}") + public static synchronized Collection<ImmutableStorageScheme> data() { + return Arrays.asList(ImmutableStorageScheme.ONE_CELL_PER_COLUMN, + ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS); + } + + private boolean isSingleCell() { + return storageScheme == ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS; + } + + /** + * Creates the base table and the vector index with the storage scheme of the test parameter, and + * activates the index with known centroids. + */ + private void setup(Connection conn, String tableName, String indexName) throws Exception { + setupTableAndKnownCentroids(conn, tableName, indexName, "FLOAT", + isSingleCell() + ? "IMMUTABLE_ROWS=true, IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS," + + " COLUMN_ENCODED_BYTES=2" + : "IMMUTABLE_STORAGE_SCHEME=ONE_CELL_PER_COLUMN, COLUMN_ENCODED_BYTES=0"); + } + + /** + * Replaces a data row. For an immutable single cell table, the method deletes the prior row + * first. Immutable index maintenance does not read the prior row, so the delete must remove the + * old index row. + */ + private void replaceRow(Connection conn, String tableName, String id, Float[] vector, + String label) throws SQLException { + if (isSingleCell()) { + try (PreparedStatement ps = + conn.prepareStatement("DELETE FROM " + tableName + " WHERE ID = ?")) { + ps.setString(1, id); + ps.executeUpdate(); + } + conn.commit(); + } + try (PreparedStatement ps = + conn.prepareStatement("UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)")) { + ps.setString(1, id); + ps.setArray(2, conn.createArrayOf("FLOAT", vector)); + ps.setString(3, label); + ps.executeUpdate(); + } + conn.commit(); + } + + /** + * Reads the serialized vector of an index row. The method supports the one cell per column and + * the single cell storage schemes. It returns null if the index row has no vector value. + */ + private byte[] getIndexedVector(PhoenixConnection pconn, PTable indexTable, byte[] rowKey) + throws Exception { + PColumn vectorCol = indexTable.getColumnForColumnName("0:V"); + try ( + Table hTable = pconn.getQueryServices().getTable(indexTable.getPhysicalName().getBytes())) { + Result r = hTable.get(new Get(rowKey)); + if (indexTable.getImmutableStorageScheme() == ImmutableStorageScheme.ONE_CELL_PER_COLUMN) { + return r.getValue(vectorCol.getFamilyName().getBytes(), + vectorCol.getColumnQualifierBytes()); + } + ImmutableBytesPtr ptr = new ImmutableBytesPtr(); + SingleCellColumnExpression expr = + new SingleCellColumnExpression(vectorCol, vectorCol.getName().getString(), + indexTable.getEncodingScheme(), indexTable.getImmutableStorageScheme()); + return expr.evaluate(new ResultTuple(r), ptr) && ptr.getLength() > 0 + ? ptr.copyBytesIfNecessary() + : null; + } + } + @Test public void testVectorInsertGeneratesIndexRow() throws Exception { String tableName = "T_VEC_INS_" + generateUniqueName(); String indexName = "IDX_VEC_INS_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { - setupTableAndKnownCentroids(conn, tableName, indexName); + setup(conn, tableName, indexName); PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); PTable indexTable = pconn.getTableNoCache(indexName); @@ -96,6 +184,9 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { List<byte[]> rowKeys = getHBaseRowKeys(pconn, indexTable); assertEquals("Expected exactly 1 index row", 1, rowKeys.size()); assertEquals("Centroid prefix must be 2", 2, extractCentroidId(rowKeys.get(0))); + assertEquals(storageScheme, indexTable.getImmutableStorageScheme()); + assertArrayEquals(PVectorFloat.INSTANCE.toBytes(new float[] { 1.0f, 0.0f, 0.0f, 0.0f }), + getIndexedVector(pconn, indexTable, rowKeys.get(0))); // Query the index table with SQL String selectSql = "SELECT \"_CENTROID_ID\", \":ID\", \"0:LABEL\" FROM " + indexName; @@ -115,7 +206,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { String indexName = "IDX_VEC_UPD_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { - setupTableAndKnownCentroids(conn, tableName, indexName); + setup(conn, tableName, indexName); PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); PTable indexTable = pconn.getTableNoCache(indexName); @@ -134,14 +225,8 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { assertEquals(1, rowKeysBefore.size()); assertEquals(2, extractCentroidId(rowKeysBefore.get(0))); - // Update vector across centroid boundary to centroid 0 - try (PreparedStatement ps = conn.prepareStatement(upsertSql)) { - ps.setString(1, "row_1"); - ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 0.0f, 0.0f, 0.0f, 1.0f })); - ps.setString(3, "lbl_updated"); - ps.executeUpdate(); - } - conn.commit(); + // Change the vector to [0,0,0,1]. The nearest centroid is ID 0. + replaceRow(conn, tableName, "row_1", new Float[] { 0.0f, 0.0f, 0.0f, 1.0f }, "lbl_updated"); // The update deletes the old index row and writes a new row under centroid 0 List<byte[]> rowKeysAfter = getHBaseRowKeys(pconn, indexTable); @@ -167,7 +252,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { String indexName = "IDX_VEC_COV_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { - setupTableAndKnownCentroids(conn, tableName, indexName); + setup(conn, tableName, indexName); PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); PTable indexTable = pconn.getTableNoCache(indexName); @@ -186,15 +271,8 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { assertEquals(1, rowKeysBefore.size()); assertEquals(2, extractCentroidId(rowKeysBefore.get(0))); - // Update non-vector covered column in place - try (PreparedStatement ps = - conn.prepareStatement("UPSERT INTO " + tableName + " (ID, V, LABEL) VALUES (?, ?, ?)")) { - ps.setString(1, "row_1"); - ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f })); - ps.setString(3, "updated_label"); - ps.executeUpdate(); - } - conn.commit(); + // Change only the covered column. The vector does not change. + replaceRow(conn, tableName, "row_1", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }, "updated_label"); // The index row key and its centroid do not change List<byte[]> rowKeysAfter = getHBaseRowKeys(pconn, indexTable); @@ -219,7 +297,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { String indexName = "IDX_VEC_DEL_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { - setupTableAndKnownCentroids(conn, tableName, indexName); + setup(conn, tableName, indexName); PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); PTable indexTable = pconn.getTableNoCache(indexName); @@ -262,7 +340,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { String indexName = "IDX_VEC_NULL_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { - setupTableAndKnownCentroids(conn, tableName, indexName); + setup(conn, tableName, indexName); PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); PTable indexTable = pconn.getTableNoCache(indexName); @@ -293,6 +371,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { @Test public void testVectorUnchangedUpdateMaintainsIndexRowAndCoveredColumns() throws Exception { + assumeFalse(isSingleCell()); String tableName = "T_VEC_ZUPD_" + generateUniqueName(); String indexName = "IDX_VEC_ZUPD_" + generateUniqueName(); @@ -369,13 +448,67 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { } } + /** + * Verifies that an update of a covered column keeps the indexed vector in a single cell index + * row. The base table uses one cell per column, and the upsert does not include the vector + * column. + */ + @Test + public void testUnchangedVectorCoveredUpdateOnSingleCellIndex() throws Exception { + assumeFalse(isSingleCell()); + String tableName = "T_VEC_SC_" + generateUniqueName(); + String indexName = "IDX_VEC_SC_" + generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + tableName + + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 4), LABEL VARCHAR)"); + stmt.execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) " + + "INCLUDE (LABEL) WITH (algorithm = 'IVF', metric = 'L2', lists = 4, sample_size = 100," + + " IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2)"); + } + VectorIndexTestUtil.activateWithKnownCentroids(conn, tableName, indexName, + VectorIndexTestUtil.KNOWN_CENTROIDS); + PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); + PTable indexTable = pconn.getTableNoCache(indexName); + assertEquals(ImmutableStorageScheme.ONE_CELL_PER_COLUMN, + pconn.getTableNoCache(tableName).getImmutableStorageScheme()); + assertEquals(ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS, + indexTable.getImmutableStorageScheme()); + + replaceRow(conn, tableName, "row_sc_1", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }, + "lbl_initial"); + try (Statement stmt = conn.createStatement()) { + stmt + .execute("UPSERT INTO " + tableName + " (ID, LABEL) VALUES ('row_sc_1', 'lbl_updated')"); + } + conn.commit(); + + List<byte[]> rowKeys = getHBaseRowKeys(pconn, indexTable); + assertEquals(1, rowKeys.size()); + assertEquals(2, extractCentroidId(rowKeys.get(0))); + assertArrayEquals(PVectorFloat.INSTANCE.toBytes(new float[] { 1.0f, 0.0f, 0.0f, 0.0f }), + getIndexedVector(pconn, indexTable, rowKeys.get(0))); + String querySql = + "SELECT ID, LABEL FROM " + tableName + " ORDER BY L2_DISTANCE(V, ?) LIMIT 1"; + try (PreparedStatement ps = conn.prepareStatement(querySql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", new Float[] { 0.9f, 0.0f, 0.0f, 0.0f })); + try (ResultSet rs = ps.executeQuery()) { + assertTrue(rs.next()); + assertEquals("row_sc_1", rs.getString(1)); + assertEquals("lbl_updated", rs.getString(2)); + assertFalse(rs.next()); + } + } + } + } + @Test public void testVerifiedMarkerPresentAfterCommit() throws Exception { String tableName = "T_VEC_VER_" + generateUniqueName(); String indexName = "IDX_VEC_VER_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { - setupTableAndKnownCentroids(conn, tableName, indexName); + setup(conn, tableName, indexName); PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); PTable indexTable = pconn.getTableNoCache(indexName); @@ -419,6 +552,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { */ @Test public void testReadRepairWithCentroidReassignment() throws Exception { + assumeFalse(isSingleCell()); String tableName = "T_VEC_RR_" + generateUniqueName(); String indexName = "IDX_VEC_RR_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { @@ -492,6 +626,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { */ @Test public void testCoveredDoubleVectorColumn() throws Exception { + assumeFalse(isSingleCell()); String tableName = "T_VEC_COVD_" + generateUniqueName(); String indexName = "IDX_VEC_COVD_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { @@ -545,6 +680,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { /** Tests index maintenance of a DOUBLE vector across insert, centroid change, and delete. */ @Test public void testDoubleVectorIndexMaintenance() throws Exception { + assumeFalse(isSingleCell()); String tableName = "T_VEC_DBL_" + generateUniqueName(); String indexName = "IDX_VEC_DBL_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { @@ -586,6 +722,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { */ @Test public void testImmutableTableClientMaintenanceMatchesServerBuild() throws Exception { + assumeFalse(isSingleCell()); String tableName = "T_VEC_IMM_" + generateUniqueName(); String indexName = "IDX_VEC_IMM_" + generateUniqueName(); try (Connection conn = DriverManager.getConnection(getUrl())) { 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 2506f615e5..0beba290b9 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 @@ -2084,6 +2084,149 @@ public class VectorIvfQueryIT extends ParallelStatsDisabledIT { } } + @Test + public void testSingleCellVectorIndexUsed() throws Exception { + String tableName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + tableName + + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 4)) " + + "IMMUTABLE_ROWS=true, IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + stmt.execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) " + + "WITH (algorithm = 'IVF', metric = 'L2', lists = 2, sample_size = 100)"); + } + // The vector index does not specify a storage scheme, so it gets the immutable storage + // scheme of the base table. + assertEquals(PTable.ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS, conn + .unwrap(PhoenixConnection.class).getTableNoCache(indexName).getImmutableStorageScheme()); + + List<float[]> centroids = Arrays.asList(new float[] { 0.0f, 0.0f, 0.0f, 0.0f }, + new float[] { 10.0f, 0.0f, 0.0f, 0.0f }); + VectorIndexTestUtil.activateWithKnownCentroids(conn, tableName, indexName, centroids); + + Map<String, float[]> rows = new LinkedHashMap<>(); + rows.put("A1", new float[] { 1.0f, 0.0f, 0.0f, 0.0f }); + rows.put("A2", new float[] { 2.0f, 0.0f, 0.0f, 0.0f }); + rows.put("B1", new float[] { 9.0f, 0.0f, 0.0f, 0.0f }); + rows.put("B2", new float[] { 10.0f, 0.0f, 0.0f, 0.0f }); + + String upsertSql = "UPSERT INTO " + tableName + " (ID, V) VALUES (?, ?)"; + try (PreparedStatement ps = conn.prepareStatement(upsertSql)) { + for (Map.Entry<String, float[]> entry : rows.entrySet()) { + ps.setString(1, entry.getKey()); + float[] v = entry.getValue(); + ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { v[0], v[1], v[2], v[3] })); + ps.executeUpdate(); + } + conn.commit(); + } + + float[] queryVec = new float[] { 0.5f, 0.0f, 0.0f, 0.0f }; + Float[] boxedQ = new Float[] { queryVec[0], queryVec[1], queryVec[2], queryVec[3] }; + int k = 2; + List<String> expectedTopK = VectorIndexTestUtil.bruteForceTopK(rows, queryVec, "L2", k); + + String querySql = "SELECT ID FROM " + tableName + " ORDER BY L2_DISTANCE(V, ?) LIMIT " + k; + try (PreparedStatement ps = conn.prepareStatement("EXPLAIN " + querySql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxedQ)); + try (ResultSet rs = ps.executeQuery()) { + String plan = QueryUtil.getExplainPlan(rs); + assertTrue("EXPLAIN plan must reference index " + indexName + ": " + plan, + plan.contains(indexName)); + assertTrue("EXPLAIN plan must contain CLIENT PROBING: " + plan, + plan.contains("CLIENT PROBING")); + } + } + + List<String> actualIds = new ArrayList<>(); + try (PreparedStatement ps = conn.prepareStatement(querySql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxedQ)); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + actualIds.add(rs.getString(1)); + } + } + } + assertEquals("Returned IDs must match brute-force top-k", expectedTopK, actualIds); + } + } + + @Test + public void testSingleCellVectorIndexWithCoveredVectorColumn() throws Exception { + String tableName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + try (Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE " + tableName + + " (ID VARCHAR NOT NULL PRIMARY KEY, V VECTOR(FLOAT, 4), COV_V VECTOR(FLOAT, 4)) " + + "IMMUTABLE_ROWS=true, IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + stmt + .execute("CREATE VECTOR INDEX " + indexName + " ON " + tableName + " (V) INCLUDE (COV_V) " + + "WITH (algorithm = 'IVF', metric = 'L2', lists = 2, sample_size = 100)"); + } + + List<float[]> centroids = Arrays.asList(new float[] { 0.0f, 0.0f, 0.0f, 0.0f }, + new float[] { 10.0f, 0.0f, 0.0f, 0.0f }); + VectorIndexTestUtil.activateWithKnownCentroids(conn, tableName, indexName, centroids); + + Map<String, float[]> rowsV = new LinkedHashMap<>(); + rowsV.put("A1", new float[] { 1.0f, 0.0f, 0.0f, 0.0f }); + rowsV.put("A2", new float[] { 2.0f, 0.0f, 0.0f, 0.0f }); + rowsV.put("B1", new float[] { 9.0f, 0.0f, 0.0f, 0.0f }); + rowsV.put("B2", new float[] { 10.0f, 0.0f, 0.0f, 0.0f }); + + Map<String, float[]> rowsCovV = new LinkedHashMap<>(); + rowsCovV.put("A1", new float[] { 100.0f, 0.0f, 0.0f, 0.0f }); + rowsCovV.put("A2", new float[] { 90.0f, 0.0f, 0.0f, 0.0f }); + rowsCovV.put("B1", new float[] { 20.0f, 0.0f, 0.0f, 0.0f }); + rowsCovV.put("B2", new float[] { 10.0f, 0.0f, 0.0f, 0.0f }); + + String upsertSql = "UPSERT INTO " + tableName + " (ID, V, COV_V) VALUES (?, ?, ?)"; + try (PreparedStatement ps = conn.prepareStatement(upsertSql)) { + for (String id : rowsV.keySet()) { + ps.setString(1, id); + float[] v = rowsV.get(id); + float[] covV = rowsCovV.get(id); + ps.setArray(2, conn.createArrayOf("FLOAT", new Float[] { v[0], v[1], v[2], v[3] })); + ps.setArray(3, + conn.createArrayOf("FLOAT", new Float[] { covV[0], covV[1], covV[2], covV[3] })); + ps.executeUpdate(); + } + conn.commit(); + } + + float[] queryVec = new float[] { 0.5f, 0.0f, 0.0f, 0.0f }; + Float[] boxedQ = new Float[] { queryVec[0], queryVec[1], queryVec[2], queryVec[3] }; + int k = 2; + List<String> expectedTopKOverV = VectorIndexTestUtil.bruteForceTopK(rowsV, queryVec, "L2", k); + + String querySql = "SELECT ID FROM " + tableName + " ORDER BY L2_DISTANCE(V, ?) LIMIT " + k; + try (PreparedStatement ps = conn.prepareStatement("EXPLAIN " + querySql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxedQ)); + try (ResultSet rs = ps.executeQuery()) { + String plan = QueryUtil.getExplainPlan(rs); + assertTrue("EXPLAIN plan must reference index " + indexName + ": " + plan, + plan.contains(indexName)); + assertTrue("EXPLAIN plan must contain CLIENT PROBING: " + plan, + plan.contains("CLIENT PROBING")); + } + } + + List<String> actualIds = new ArrayList<>(); + try (PreparedStatement ps = conn.prepareStatement(querySql)) { + ps.setArray(1, conn.createArrayOf("FLOAT", boxedQ)); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + actualIds.add(rs.getString(1)); + } + } + } + assertEquals("Returned IDs must match brute force over indexed column V", expectedTopKOverV, + actualIds); + } + } + /** * Tests writes and a vector top-K query through a view with a WHERE clause. The view inherits the * vector index of the table under a view-specific name. That inherited index must assign and 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 29166cd53a..138700113c 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 @@ -657,6 +657,45 @@ public class IndexMaintainerTest extends BaseConnectionlessQueryTest { } } + /** + * Verifies that a single cell index row holds the correct covered column value when the data + * table and the index use different qualifier encoding schemes. + */ + @Test + public void testSingleCellCoveredValueAcrossEncodingSchemes() 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, K VARCHAR, C VARCHAR) IMMUTABLE_ROWS=true," + + " IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + pconn.createStatement().execute("CREATE INDEX " + indexName + " ON " + tableName + + " (K) INCLUDE (C) COLUMN_ENCODED_BYTES=4"); + PTable dataTable = pconn.getTable(tableName); + PTable index = pconn.getTable(indexName); + assertEquals(PTable.QualifierEncodingScheme.TWO_BYTE_QUALIFIERS, + dataTable.getEncodingScheme()); + assertEquals(PTable.QualifierEncodingScheme.FOUR_BYTE_QUALIFIERS, index.getEncodingScheme()); + IndexMaintainer im = index.getIndexMaintainer(dataTable, pconn); + + Put dataRow = dataRow(pconn, "UPSERT INTO " + tableName + " VALUES ('a', 'k', 'covered')"); + Put indexPut = im.buildUpdateMutation(GenericKeyValueBuilder.INSTANCE, + new IndexUtil.SimpleValueGetter(dataRow), new ImmutableBytesPtr(dataRow.getRow()), + EnvironmentEdgeManager.currentTimeMillis(), null, null, false, null); + PColumn indexCol = index.getColumnForColumnName("0:C"); + List<Cell> cells = indexPut.get(indexCol.getFamilyName().getBytes(), + QueryConstants.SINGLE_KEYVALUE_COLUMN_QUALIFIER_BYTES); + assertEquals(1, cells.size()); + ImmutableBytesWritable ptr = new ImmutableBytesWritable(CellUtil.cloneValue(cells.get(0))); + assertTrue(index.getImmutableStorageScheme().getDecoder().decode(ptr, + index.getEncodingScheme().decode(indexCol.getColumnQualifierBytes()) + - QueryConstants.ENCODED_CQ_COUNTER_INITIAL_VALUE + 1)); + assertEquals("covered", Bytes.toString(ptr.copyBytes())); + } + } + @Test public void testNonVectorIndexMaintainerHasNoVectorFields() throws Exception { String tableName = "T_" + generateUniqueName();
