CNDB-14773: avoid Int2IntHashMap overflow in RAMStringIndexer and improve memory track to include array memory usage (#1885)

### What is the issue
#14773: Int2IntHashMap overflow when num of docs reaches 348_966_081

### What does this PR fix and why was it fixed

Trigger segment flush before overflow and include array memory usage to
avoid undercounting memory usage

Author: jasonstack

src/java/org/apache/cassandra/index/sai/disk/RAMPostingSlices.java

@@ -49,6 +49,11 @@ class RAMPostingSlices
         this.includeFrequencies = includeFrequencies;
     }
 
+    long arrayMemoryUsage()
+    {
+        return postingStarts.length * 4L + postingUptos.length * 4L + sizes.length * 4L;
+    }
+
     /**
      * Creates and returns a PostingList for the given term ID.
      */

src/java/org/apache/cassandra/index/sai/disk/RAMStringIndexer.java

@@ -39,6 +39,15 @@ public class RAMStringIndexer
 {
     @VisibleForTesting
     public static int MAX_BLOCK_BYTE_POOL_SIZE = Integer.MAX_VALUE;
+
+    /**
+     * Int2IntHashMap "docLengths" needs to resize when size reaches 348_966_081 (capacity * loadFactor). At that point, its capacity is 536870912.
+     * Its new capacity will be quadrupled and exceed Integer.MAX_VALUE.
+     *
+     * Pick 300_000_000 for simplicity to trigger segment flush.
+     */
+    private static final int MAX_DOCS_SIZE = 300_000_000;
+
     private final BytesRefHash termsHash;
     private final RAMPostingSlices slices;
     // counters need to be separate so that we can trigger flushes if either ByteBlockPool hits maximum size
@@ -48,11 +57,19 @@ public class RAMStringIndexer
     private int[] lastSegmentRowID = new int[RAMPostingSlices.DEFAULT_TERM_DICT_SIZE];
 
     private final boolean writeFrequencies;
+    private final int maxDocSize;
     private final Int2IntHashMap docLengths = new Int2IntHashMap(Integer.MIN_VALUE);
 
     public RAMStringIndexer(boolean writeFrequencies)
+    {
+        this(writeFrequencies, MAX_DOCS_SIZE);
+    }
+
+    @VisibleForTesting
+    RAMStringIndexer(boolean writeFrequencies, int maxDocSize)
     {
         this.writeFrequencies = writeFrequencies;
+        this.maxDocSize = maxDocSize;
         termsBytesUsed = Counter.newCounter();
         slicesBytesUsed = Counter.newCounter();
 
@@ -65,7 +82,11 @@ public RAMStringIndexer(boolean writeFrequencies)
 
     public long estimatedBytesUsed()
     {
-        return termsBytesUsed.get() + slicesBytesUsed.get();
+        // record the array memory usage from Int2IntHashMap docLengths:
+        //  * array size is capacity * 2
+        //  * 4 bytes per int
+        long docLengthsMemoryUsage = docLengths.capacity() * 2 * 4L;
+        return docLengthsMemoryUsage + termsBytesUsed.get() + slicesBytesUsed.get() + slices.arrayMemoryUsage();
     }
 
     public boolean requiresFlush()
@@ -75,7 +96,9 @@ public boolean requiresFlush()
         // be triggered by an addition, and the rest of the space in the final chunk will be wasted, as the bytesUsed
         // counters track block allocation, not the size of additions. This means that we can't pass this check and then
         // fail to add a term.
-        return termsBytesUsed.get() >= MAX_BLOCK_BYTE_POOL_SIZE || slicesBytesUsed.get() >= MAX_BLOCK_BYTE_POOL_SIZE;
+        return termsBytesUsed.get() >= MAX_BLOCK_BYTE_POOL_SIZE || slicesBytesUsed.get() >= MAX_BLOCK_BYTE_POOL_SIZE
+               // to avoid Int2IntHashMap new capacity overflow
+               || docLengths.size() >= maxDocSize;
     }
 
     public boolean isEmpty()

test/unit/org/apache/cassandra/index/sai/disk/RAMStringIndexerTest.java

@@ -77,6 +77,26 @@ public void test() throws Exception
         }
     }
 
+    @Test
+    public void testLargeNumberOfDocs()
+    {
+        int maxDocsSize = 1000;
+        RAMStringIndexer indexer = new RAMStringIndexer(false, maxDocsSize);
+
+        int startingRowId = 0;
+        int i = 0;
+        while (i++ < maxDocsSize)
+        {
+            int rowId = startingRowId + i;
+            indexer.addAll(List.of(new BytesRef("0")), rowId);
+
+            if (i < maxDocsSize)
+                assertFalse(indexer.requiresFlush());
+        }
+
+        assertTrue(indexer.requiresFlush());
+    }
+
     @Test
     public void testWithFrequencies() throws Exception
     {