CNDB-13696 fix empty iterator access in BM25 search on partial SSTable (#1691)

Executing a query with BM25 search and a condition on partial SSTable
results in empty iterator access error. And there was no test with
storing data in segments.

The PR implements BM25 search tests with splitting data into two tables.
This reproduced this bug, CNDB-13696, and demonstrates current confusion
on the BM25 ordering result to be fixed by CNDB-13553.

This PR adds a check for empty iterator created for a PK belonging to
another segment. This fixes the bug of trying to get the first element
of an empty iterator.

Author: k-rus

src/java/org/apache/cassandra/index/sai/disk/v1/InvertedIndexSearcher.java

@@ -168,6 +168,9 @@ private Cell<?> readColumn(SSTableReader sstable, PrimaryKey primaryKey)
         var slices = Slices.with(indexContext.comparator(), Slice.make(primaryKey.clustering()));
         try (var rowIterator = sstable.rowIterator(dk, slices, columnFilter, false, SSTableReadsListener.NOOP_LISTENER))
         {
+            // primaryKey might not belong to this sstable, thus the iterator will be empty
+            if (rowIterator.isEmpty())
+                return null;
             var unfiltered = rowIterator.next();
             assert unfiltered.isRow() : unfiltered;
             Row row = (Row) unfiltered;

test/unit/org/apache/cassandra/index/sai/cql/BM25Test.java

@@ -749,6 +749,42 @@ public void testCollections() throws Throwable
         });
     }
 
+    @Test
+    public void testOrderingSeveralSegments() throws Throwable
+    {
+        createTable("CREATE TABLE %s (id int PRIMARY KEY, category text, score int," +
+                    "title text, body text)");
+        createAnalyzedIndex("body", true);
+        createIndex("CREATE CUSTOM INDEX ON %s (score) USING 'StorageAttachedIndex'");
+        insertPrimitiveData(0, 10);
+        flush();
+        insertPrimitiveData(10, 20);
+
+        // One memtable, one sstable - different result from the reference in testCollections
+        // ID 1 and 6 contain 3 and 2 climate occurrences correspondingly,
+        // while ID 11 and 19 - 4 climate occurrences. However,
+        // since the segment with 0-9 IDs have only 2 rows with climate and 10-19 - 5,
+        // 1 and 6 win over 11 and 19.
+        executeQuery(Arrays.asList(1, 6, 11, 19, 16, 12, 18), "SELECT * FROM %s  ORDER BY body BM25 OF ? LIMIT 10",
+                "climate");
+        executeQuery(Arrays.asList(1, 11, 19), "SELECT * FROM %s WHERE score = 5 ORDER BY body BM25 OF ? LIMIT 10",
+                "climate");
+
+        // Flush into Two sstables - same result as the different above
+        flush();
+        executeQuery(Arrays.asList(1, 6, 11, 19, 16, 12, 18), "SELECT * FROM %s  ORDER BY body BM25 OF ? LIMIT 10",
+                "climate");
+        executeQuery(Arrays.asList(1, 11, 19), "SELECT * FROM %s WHERE score = 5 ORDER BY body BM25 OF ? LIMIT 10",
+                "climate");
+
+        // Compact into one sstable - same as reference from testCollections
+        compact();
+        executeQuery(Arrays.asList(11, 19, 1, 16, 6, 12, 18), "SELECT * FROM %s  ORDER BY body BM25 OF ? LIMIT 10",
+                "climate");
+        executeQuery(Arrays.asList(11, 19, 1), "SELECT * FROM %s WHERE score = 5 ORDER BY body BM25 OF ? LIMIT 10",
+                "climate");
+    }
+
     private final static Object[][] DATASET =
     {
     { 1, "Climate", 5, "Climate change is a pressing issue. Climate patterns are shifting globally. Scientists study climate data daily.", 1 },
@@ -794,8 +830,14 @@ private void analyzeDataset(String term)
 
     private void insertPrimitiveData()
     {
-        for (Object[] row : DATASET)
+        insertPrimitiveData(0, DATASET.length);
+    }
+
+    private void insertPrimitiveData(int start, int end)
+    {
+        for (int i = start; i < end; i++)
         {
+            Object[] row = DATASET[i];
             execute(
             "INSERT INTO %s (id, category, score, body) VALUES (?, ?, ?, ?)",
             row[0],