CNDB-13480: NullPointerException when using ORDER BY BM25 after deleting a row (#1653)

### What is the issue
see CNDB-13480


### What does this PR fix and why was it fixed
Exclude deleted rows while computing terms frequencies and preparing for
computing the scores

Author: eolivelli

src/java/org/apache/cassandra/index/sai/memory/TrieMemtableIndex.java

@@ -45,6 +45,7 @@
 import org.apache.cassandra.db.memtable.Memtable;
 import org.apache.cassandra.db.memtable.ShardBoundaries;
 import org.apache.cassandra.db.memtable.TrieMemtable;
+import org.apache.cassandra.db.rows.Cell;
 import org.apache.cassandra.db.rows.Row;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.Bounds;
@@ -266,7 +267,15 @@ public List<CloseableIterator<PrimaryKeyWithSortKey>> orderBy(QueryContext query
         // Compute BM25 scores
         var docStats = computeDocumentFrequencies(queryContext, queryTerms);
         var analyzer = indexContext.getAnalyzerFactory().create();
-        var it = Iterators.transform(intersectedIterator, pk -> BM25Utils.DocTF.createFromDocument(pk, getCellForKey(pk), analyzer, queryTerms));
+        var it = Iterators.filter(Iterators.transform(intersectedIterator, pk ->
+        {
+            Cell<?> cellForKey = getCellForKey(pk);
+            if (cellForKey == null)
+                // skip deleted rows
+                return null;
+            return BM25Utils.DocTF.createFromDocument(pk, cellForKey, analyzer, queryTerms);
+        }), Objects::nonNull);
+
         return List.of(BM25Utils.computeScores(CloseableIterator.wrap(it),
                                                queryTerms,
                                                docStats,
@@ -331,7 +340,13 @@ public CloseableIterator<PrimaryKeyWithSortKey> orderResultsBy(QueryContext quer
         var analyzer = indexContext.getAnalyzerFactory().create();
         var queryTerms = orderer.getQueryTerms();
         var docStats = computeDocumentFrequencies(queryContext, queryTerms);
-        var it = keys.stream().map(pk -> BM25Utils.DocTF.createFromDocument(pk, getCellForKey(pk), analyzer, queryTerms)).iterator();
+        var it = keys.stream().map(pk -> {
+            Cell<?> cellForKey = getCellForKey(pk);
+            if (cellForKey == null)
+                // skip deleted rows
+                return null;
+            return BM25Utils.DocTF.createFromDocument(pk, cellForKey, analyzer, queryTerms);
+        }).filter(Objects::nonNull).iterator();
         return BM25Utils.computeScores(CloseableIterator.wrap(it),
                                        queryTerms,
                                        docStats,
@@ -350,8 +365,15 @@ private BM25Utils.DocStats computeDocumentFrequencies(QueryContext queryContext,
         {
             // KeyRangeIterator.getMaxKeys is not accurate enough, we have to count them
             long keys = 0;
-            for (var it = termIterators.get(i); it.hasNext(); it.next())
+            for (var it = termIterators.get(i); it.hasNext(); )
+            {
+                PrimaryKey pk = it.next();
+                Cell<?> cellForKey = getCellForKey(pk);
+                if (cellForKey == null)
+                    // skip deleted rows
+                    continue;
                 keys++;
+            }
             documentFrequencies.put(queryTerms.get(i), keys);
         }
         long docCount = 0;

src/java/org/apache/cassandra/index/sai/utils/BM25Utils.java

@@ -83,9 +83,10 @@ public static DocTF createFromDocument(PrimaryKey pk,
                                              AbstractAnalyzer docAnalyzer,
                                              Collection<ByteBuffer> queryTerms)
         {
+            assert cell != null : "Cannot find a cell for pk " + pk;
+
             int count = 0;
             Map<ByteBuffer, Integer> frequencies = new HashMap<>();
-
             docAnalyzer.reset(cell.buffer());
             try
             {

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

@@ -57,20 +57,60 @@ public void testTwoIndexes()
         assertInvalidMessage("BM25 ordering on column v requires an analyzed index",
                              "SELECT k FROM %s WHERE v : 'apple' ORDER BY v BM25 OF 'apple' LIMIT 3");
 
-        // create analyzed index
-        analyzeIndex();
+        createAnalyzedIndex();
         // BM25 query should work now
         var result = execute("SELECT k FROM %s WHERE v : 'apple' ORDER BY v BM25 OF 'apple' LIMIT 3");
         assertRows(result, row(1));
     }
 
+    @Test
+    public void testDeletedRow() throws Throwable
+    {
+        createTable("CREATE TABLE %s (k int PRIMARY KEY, v text)");
+        createAnalyzedIndex();
+        execute("INSERT INTO %s (k, v) VALUES (1, 'apple')");
+        execute("INSERT INTO %s (k, v) VALUES (2, 'apple juice')");
+        var result = execute("SELECT k FROM %s ORDER BY v BM25 OF 'apple' LIMIT 3");
+        assertThat(result).hasSize(2);
+        execute("DELETE FROM %s WHERE k=2");
+        String select = "SELECT k FROM %s ORDER BY v BM25 OF 'apple' LIMIT 3";
+        beforeAndAfterFlush(() -> assertRows(execute(select), row(1)));
+    }
+
+    @Test
+    public void testDeletedColumn() throws Throwable
+    {
+        createTable("CREATE TABLE %s (k int PRIMARY KEY, v text)");
+        createAnalyzedIndex();
+        execute("INSERT INTO %s (k, v) VALUES (1, 'apple')");
+        execute("INSERT INTO %s (k, v) VALUES (2, 'apple juice')");
+        String select = "SELECT k FROM %s ORDER BY v BM25 OF 'apple' LIMIT 3";
+        assertRows(execute(select), row(1), row(2));
+        execute("DELETE v FROM %s WHERE k = 2");
+        beforeAndAfterFlush(() -> assertRows(execute(select), row(1)));
+    }
+
+    @Test
+    public void testDeletedRowWithPredicate() throws Throwable
+    {
+        createTable("CREATE TABLE %s (k int PRIMARY KEY, v text, n int)");
+        createIndex("CREATE CUSTOM INDEX ON %s(n) USING 'org.apache.cassandra.index.sai.StorageAttachedIndex'");
+        createAnalyzedIndex();
+        execute("INSERT INTO %s (k, v, n) VALUES (1, 'apple', 0)");
+        execute("INSERT INTO %s (k, v, n) VALUES (2, 'apple juice', 0)");
+        String select = "SELECT k FROM %s WHERE n = 0 ORDER BY v BM25 OF 'apple' LIMIT 3";
+        assertRows(execute(select), row(1), row(2));
+        execute("DELETE FROM %s WHERE k=2");
+        beforeAndAfterFlush(() -> assertRows(execute(select), row(1)));
+    }
+
     @Test
     public void testTwoIndexesAmbiguousPredicate() throws Throwable
     {
         createTable("CREATE TABLE %s (k int PRIMARY KEY, v text)");
 
-        // Create analyzed and un-analyzed indexes
-        analyzeIndex();
+        createAnalyzedIndex();
+        // Create  un-analyzed indexes
         createIndex("CREATE CUSTOM INDEX ON %s(v) USING 'org.apache.cassandra.index.sai.StorageAttachedIndex'");
 
         execute("INSERT INTO %s (k, v) VALUES (1, 'apple')");
@@ -368,10 +408,10 @@ public void testIrrelevantRowsWithCompaction()
     private void createSimpleTable()
     {
         createTable("CREATE TABLE %s (k int PRIMARY KEY, v text)");
-        analyzeIndex();
+        createAnalyzedIndex();
     }
 
-    private String analyzeIndex()
+    private String createAnalyzedIndex()
     {
         return createIndex("CREATE CUSTOM INDEX ON %s(v) " +
                            "USING 'org.apache.cassandra.index.sai.StorageAttachedIndex' " +
@@ -387,7 +427,7 @@ private String analyzeIndex()
     public void testWithPredicate() throws Throwable
     {
         createTable("CREATE TABLE %s (k int PRIMARY KEY, p int, v text)");
-        analyzeIndex();
+        createAnalyzedIndex();
         execute("CREATE CUSTOM INDEX ON %s(p) USING 'StorageAttachedIndex'");
 
         // Insert documents with varying frequencies of the term "apple"
@@ -412,7 +452,7 @@ public void testWithPredicate() throws Throwable
     public void testWidePartition() throws Throwable
     {
         createTable("CREATE TABLE %s (k1 int, k2 int, v text, PRIMARY KEY (k1, k2))");
-        analyzeIndex();
+        createAnalyzedIndex();
 
         // Insert documents with varying frequencies of the term "apple"
         execute("INSERT INTO %s (k1, k2, v) VALUES (0, 1, 'apple')");
@@ -434,7 +474,7 @@ public void testWidePartition() throws Throwable
     public void testWidePartitionWithPkPredicate() throws Throwable
     {
         createTable("CREATE TABLE %s (k1 int, k2 int, v text, PRIMARY KEY (k1, k2))");
-        analyzeIndex();
+        createAnalyzedIndex();
 
         // Insert documents with varying frequencies of the term "apple"
         execute("INSERT INTO %s (k1, k2, v) VALUES (0, 1, 'apple')");
@@ -458,7 +498,7 @@ public void testWidePartitionWithPkPredicate() throws Throwable
     public void testWidePartitionWithPredicate() throws Throwable
     {
         createTable("CREATE TABLE %s (k1 int, k2 int, p int, v text, PRIMARY KEY (k1, k2))");
-        analyzeIndex();
+        createAnalyzedIndex();
         execute("CREATE CUSTOM INDEX ON %s(p) USING 'StorageAttachedIndex'");
 
         // Insert documents with varying frequencies of the term "apple"
@@ -519,7 +559,7 @@ public void testQueryEmptyTable()
     public void testBM25RaceConditionConcurrentQueriesInInvertedIndexSearcher() throws Throwable
     {
         createTable("CREATE TABLE %s (pk int, v text, PRIMARY KEY (pk))");
-        analyzeIndex();
+        createAnalyzedIndex();
 
         // Create 3 docs that have the same BM25 score and will be our top docs
         execute("INSERT INTO %s (pk, v) VALUES (1, 'apple apple apple')");
@@ -552,7 +592,7 @@ public void testBM25RaceConditionConcurrentQueriesInInvertedIndexSearcher() thro
     public void testWildcardSelection()
     {
         createTable("CREATE TABLE %s (k int, c int, v text, PRIMARY KEY (k, c))");
-        analyzeIndex();
+        createAnalyzedIndex();
         execute("INSERT INTO %s (k, c, v) VALUES (1, 1, 'apple')");
 
         var result = execute("SELECT * FROM %s ORDER BY v BM25 OF 'apple' LIMIT 3");