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 e68ff07b6ea0e9300171aa8419ed45924513a1e7 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 | 180 ++++++++++++++++++--- .../apache/phoenix/end2end/VectorIvfQueryIT.java | 142 ++++++++++++++++ .../apache/phoenix/index/IndexMaintainerTest.java | 39 +++++ 4 files changed, 340 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 d9c294edfc..782be3d807 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 @@ -1537,8 +1537,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 bb834480d1..e68665cb6b 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 write path vector index maintenance, covering centroid key assignment, @@ -68,15 +78,91 @@ import org.junit.experimental.categories.Category; * markers, read repair, and immutable client side maintenance. */ @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; + } + + /** + * Initializes the base table and associated vector index for the parameterized storage scheme. + */ + 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 immutable single cell tables, deletes the prior row before upserting + * to ensure complete index row replacement without prior row reads. + */ + 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(); + } + + /** + * Extracts the serialized vector bytes from an index row across one cell per column and single + * cell storage schemes. + */ + 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); @@ -95,6 +181,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 index table via SQL String selectSql = "SELECT \"_CENTROID_ID\", \":ID\", \"0:LABEL\" FROM " + indexName; @@ -114,7 +203,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); @@ -133,14 +222,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(); + // Update vector to [0,0,0,1] -> nearest centroid ID 0 + replaceRow(conn, tableName, "row_1", new Float[] { 0.0f, 0.0f, 0.0f, 1.0f }, "lbl_updated"); // Assert old index key deletion and new index key creation across partitions List<byte[]> rowKeysAfter = getHBaseRowKeys(pconn, indexTable); @@ -166,7 +249,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); @@ -185,15 +268,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(); + // Update ONLY covered column (same vector, new label) + replaceRow(conn, tableName, "row_1", new Float[] { 1.0f, 0.0f, 0.0f, 0.0f }, "updated_label"); // Assert index row key and centroid partition remain unchanged List<byte[]> rowKeysAfter = getHBaseRowKeys(pconn, indexTable); @@ -218,7 +294,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); @@ -261,7 +337,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); @@ -292,6 +368,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(); @@ -368,13 +445,66 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { } } + /** + * Verifies that single cell vector index rows preserve indexed vectors when updating covered + * columns on a multi-cell base table. + */ + @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); @@ -418,6 +548,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())) { @@ -488,6 +619,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { /** Preserves covered vector columns of differing element widths in index rows. */ @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())) { @@ -541,6 +673,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { /** Validates double precision vector index maintenance across mutations and boundary updates. */ @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())) { @@ -579,6 +712,7 @@ public class VectorIndexWriteIT extends ParallelStatsDisabledIT { /** Validates client side immutable index generation parity against server side builds. */ @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 3d492ac05d..47d776794c 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 @@ -1948,4 +1948,146 @@ public class VectorIvfQueryIT extends ParallelStatsDisabledIT { .contains("77.125")); } } + + @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)"); + } + // Verify vector index inherits 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); + } + } } 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 5305c2f986..baeb53f3d6 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 @@ -634,6 +634,45 @@ public class IndexMaintainerTest extends BaseConnectionlessQueryTest { } } + /** + * Verifies covered column value extraction when the data table and single cell 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();
