CNDB-15920: Add a test for disk usage table state metric after compaction

Author: ekaterinadimitrova2

test/unit/org/apache/cassandra/index/sai/functional/DiskSpaceTest.java

@@ -22,6 +22,7 @@
 import org.apache.cassandra.index.sai.SAITester;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 public class DiskSpaceTest extends SAITester
 {
@@ -52,4 +53,63 @@ public void testTableTotalDiskSpaceUsed() throws Throwable
         assertEquals(sstableSize, totalDiskSpaceUsed());
         verifyIndexComponentsNotIncludedInSSTable();
     }
+
+    @Test
+    public void testDiskUsageWithCompactionAfterUpdates() throws Throwable
+    {
+        createTable(CREATE_TABLE_TEMPLATE);
+
+        // Insert initial data and flush to create first SSTable
+        int rows = 100;
+        for (int j = 0; j < rows; j++)
+            execute("INSERT INTO %s (id1, v1) VALUES (?, ?)", Integer.toString(j), j);
+
+        flush();
+        
+        assertEquals("Should have 1 SSTable after first flush", 1, getCurrentColumnFamilyStore().getLiveSSTables().size());
+        long sstableSizeAfterFirstFlush = totalDiskSpaceUsed();
+
+        // Update the same keys with different values to create overlapping data
+        for (int j = 0; j < rows; j++)
+            execute("INSERT INTO %s (id1, v1) VALUES (?, ?)", Integer.toString(j), j + 1000);
+
+        flush();
+        
+        assertEquals("Should have 2 SSTables after second flush", 2, getCurrentColumnFamilyStore().getLiveSSTables().size());
+        long sstableSizeBeforeCompaction = totalDiskSpaceUsed();
+        
+        // Verify that we have more disk usage with 2 SSTables containing overlapping data
+        assertTrue("Disk usage should increase with overlapping SSTables",
+                   sstableSizeBeforeCompaction > sstableSizeAfterFirstFlush);
+
+        // Compact to merge the updates before creating index
+        compact();
+        waitForCompactionsFinished();
+        
+        assertEquals("Should have 1 SSTable after compaction", 1, getCurrentColumnFamilyStore().getLiveSSTables().size());
+        long sstableSizeAfterCompaction = totalDiskSpaceUsed();
+        
+        // Compaction should reduce disk usage by merging duplicate keys
+        assertTrue("Compaction should reduce disk usage by merging duplicates",
+                   sstableSizeAfterCompaction < sstableSizeBeforeCompaction);
+
+        // Create index on the compacted SSTable
+        String indexName = createIndex(String.format(CREATE_INDEX_TEMPLATE, "v1"));
+
+        long indexSize = indexDiskSpaceUse();
+        long sstableSizeWithIndex = totalDiskSpaceUsed();
+        
+        assertTrue("Index size should be positive", indexSize > 0);
+        
+        // Verify that total disk usage equals base table plus index after compaction
+        assertEquals("Total disk should equal base table plus index after compaction",
+                     sstableSizeAfterCompaction + indexSize, sstableSizeWithIndex);
+        verifyIndexComponentsIncludedInSSTable();
+
+        // Drop index and verify disk space accounting
+        dropIndex("DROP INDEX %s." + indexName);
+        assertEquals("After dropping index, disk usage should equal base table only",
+                     sstableSizeAfterCompaction, totalDiskSpaceUsed());
+        verifyIndexComponentsNotIncludedInSSTable();
+    }
 }