Merge remote-tracking branch 'datastax/main' into revive-fused-adc

Author: michaeljmarshall

src/java/org/apache/cassandra/db/marshal/AbstractType.java

@@ -37,6 +37,7 @@
 import org.apache.cassandra.cql3.CQL3Type;
 import org.apache.cassandra.cql3.ColumnSpecification;
 import org.apache.cassandra.cql3.Term;
+import org.apache.cassandra.db.rows.Cell;
 import org.apache.cassandra.exceptions.InvalidColumnTypeException;
 import org.apache.cassandra.exceptions.SyntaxException;
 import org.apache.cassandra.io.util.DataInputPlus;

src/java/org/apache/cassandra/db/virtual/TableMetricTables.java

@@ -71,9 +71,11 @@ public static Collection<VirtualTable> getAll(String name)
             new LatencyTableMetric(name, "local_read_latency", t -> t.readLatency.tableOrKeyspaceMetric().latency),
             new LatencyTableMetric(name, "local_scan_latency", t -> t.rangeLatency.tableOrKeyspaceMetric().latency),
             new LatencyTableMetric(name, "coordinator_read_latency", t -> t.coordinatorReadLatency.tableOrKeyspaceTimer()),
+            new LatencyTableMetric(name, "coordinator_cas_read_latency", t -> t.coordinatorCasReadLatency.tableOrKeyspaceTimer()),
             new LatencyTableMetric(name, "coordinator_scan_latency", t -> t.coordinatorScanLatency.tableOrKeyspaceTimer()),
             new LatencyTableMetric(name, "local_write_latency", t -> t.writeLatency.tableOrKeyspaceMetric().latency),
             new LatencyTableMetric(name, "coordinator_write_latency", t -> t.coordinatorWriteLatency.tableOrKeyspaceTimer()),
+            new LatencyTableMetric(name, "coordinator_cas_write_latency", t -> t.coordinatorCasWriteLatency.tableOrKeyspaceTimer()),
             new HistogramTableMetric(name, "tombstones_per_read", t -> t.tombstoneScannedHistogram.tableOrKeyspaceHistogram()),
             new HistogramTableMetric(name, "rows_per_read", t -> t.liveScannedHistogram.tableOrKeyspaceHistogram()),
             new StorageTableMetric(name, "disk_usage", (TableMetrics t) -> t.totalDiskSpaceUsed),

src/java/org/apache/cassandra/index/sai/IndexContext.java

@@ -614,7 +614,7 @@ public int getIntOption(String name, int defaultValue)
         }
         catch (NumberFormatException e)
         {
-            logger.error("Failed to parse index configuration " + name + " = " + value + " as integer");
+            logger.error("Failed to parse index configuration {} = {} as integer", name, value);
             return defaultValue;
         }
     }
@@ -1003,7 +1003,7 @@ public long indexFileCacheSize()
     public IndexFeatureSet indexFeatureSet()
     {
         IndexFeatureSet.Accumulator accumulator = new IndexFeatureSet.Accumulator(version);
-        getView().getIndexes().stream().map(SSTableIndex::indexFeatureSet).forEach(set -> accumulator.accumulate(set));
+        getView().getIndexes().stream().map(SSTableIndex::indexFeatureSet).forEach(accumulator::accumulate);
         return accumulator.complete();
     }
 }

src/java/org/apache/cassandra/index/sai/QueryContext.java

@@ -44,9 +44,31 @@ public class QueryContext
 
     private final LongAdder sstablesHit = new LongAdder();
     private final LongAdder segmentsHit = new LongAdder();
-    private final LongAdder partitionsRead = new LongAdder();
-    private final LongAdder rowsPreFiltered = new LongAdder();
-    private final LongAdder rowsFiltered = new LongAdder();
+
+    /**
+     * The partition/row keys that will be used to fetch rows from the base table.
+     * They will be either partition keys in AA, or row keys in the later row-aware disk formats.
+     */
+    private final LongAdder keysFetched = new LongAdder();
+
+    /** The number of live partitions fetched from the storage engine, before post-filtering. */
+    private final LongAdder partitionsFetched = new LongAdder();
+
+    /** The number of live partitions returned to the coordinator, after post-filtering. */
+    private final LongAdder partitionsReturned = new LongAdder();
+
+    /** The number of deleted partitions that are fetched. */
+    private final LongAdder partitionTombstonesFetched = new LongAdder();
+
+    /** The number of live rows fetched from the storage engine, before post-filtering. */
+    private final LongAdder rowsFetched = new LongAdder();
+
+    /** The number of live rows returned to the coordinator, after post-filtering. */
+    private final LongAdder rowsReturned = new LongAdder();
+
+    /** The number of deleted individual rows or ranges of rows that are fetched. */
+    private final LongAdder rowTombstonesFetched = new LongAdder();
+
     private final LongAdder trieSegmentsHit = new LongAdder();
 
     private final LongAdder bkdPostingListsHit = new LongAdder();
@@ -64,8 +86,6 @@ public class QueryContext
 
     private float annRerankFloor = 0.0f; // only called from single-threaded setup code
 
-    private final LongAdder shadowedPrimaryKeyCount = new LongAdder();
-
     // Determines the order of using indexes for filtering and sorting.
     // Null means the query execution order hasn't been decided yet.
     private FilterSortOrder filterSortOrder = null;
@@ -92,49 +112,81 @@ public void addSstablesHit(long val)
     {
         sstablesHit.add(val);
     }
+
     public void addSegmentsHit(long val) {
         segmentsHit.add(val);
     }
-    public void addPartitionsRead(long val)
+
+    public void addKeysFetched(long val)
+    {
+        keysFetched.add(val);
+    }
+
+    public void addPartitionsFetched(long val)
+    {
+        partitionsFetched.add(val);
+    }
+
+    public void addPartitionsReturned(long val)
     {
-        partitionsRead.add(val);
+        partitionsReturned.add(val);
     }
-    public void addRowsFiltered(long val)
+
+    public void addPartitionTombstonesFetched(long val)
     {
-        rowsFiltered.add(val);
+        partitionTombstonesFetched.add(val);
     }
-    public void addRowsPreFiltered(long val)
+
+    public void addRowsFetched(long val)
     {
-        rowsPreFiltered.add(val);
+        rowsFetched.add(val);
     }
+
+    public void addRowsReturned(long val)
+    {
+        rowsReturned.add(val);
+    }
+
+    public void addRowTombstonesFetched(long val)
+    {
+        rowTombstonesFetched.add(val);
+    }
+
     public void addTrieSegmentsHit(long val)
     {
         trieSegmentsHit.add(val);
     }
+
     public void addBkdPostingListsHit(long val)
     {
         bkdPostingListsHit.add(val);
     }
+
     public void addBkdSegmentsHit(long val)
     {
         bkdSegmentsHit.add(val);
     }
+
     public void addBkdPostingsSkips(long val)
     {
         bkdPostingsSkips.add(val);
     }
+
     public void addBkdPostingsDecodes(long val)
     {
         bkdPostingsDecodes.add(val);
     }
+
     public void addTriePostingsSkips(long val)
     {
         triePostingsSkips.add(val);
     }
+
     public void addTriePostingsDecodes(long val)
     {
         triePostingsDecodes.add(val);
     }
+
     public void addQueryTimeouts(long val)
     {
         queryTimeouts.add(val);
@@ -156,53 +208,86 @@ public long sstablesHit()
     {
         return sstablesHit.longValue();
     }
+
     public long segmentsHit() {
         return segmentsHit.longValue();
     }
-    public long partitionsRead()
+
+    public long keysFetched()
     {
-        return partitionsRead.longValue();
+        return keysFetched.longValue();
     }
-    public long rowsFiltered()
+
+    public long partitionsFetched()
     {
-        return rowsFiltered.longValue();
+        return partitionsFetched.longValue();
     }
-    public long rowsPreFiltered()
+
+    public long partitionsReturned()
+    {
+        return partitionsReturned.longValue();
+    }
+
+    public long partitionTombstonesFetched()
+    {
+        return partitionTombstonesFetched.longValue();
+    }
+
+    public long rowsFetched()
+    {
+        return rowsFetched.longValue();
+    }
+
+    public long rowsReturned()
+    {
+        return rowsReturned.longValue();
+    }
+
+    public long rowTombstonesFetched()
     {
-        return rowsPreFiltered.longValue();
+        return rowTombstonesFetched.longValue();
     }
+
     public long trieSegmentsHit()
     {
         return trieSegmentsHit.longValue();
     }
+
     public long bkdPostingListsHit()
     {
         return bkdPostingListsHit.longValue();
     }
+
     public long bkdSegmentsHit()
     {
         return bkdSegmentsHit.longValue();
     }
+
     public long bkdPostingsSkips()
     {
         return bkdPostingsSkips.longValue();
     }
+
     public long bkdPostingsDecodes()
     {
         return bkdPostingsDecodes.longValue();
     }
+
     public long triePostingsSkips()
     {
         return triePostingsSkips.longValue();
     }
+
     public long triePostingsDecodes()
     {
         return triePostingsDecodes.longValue();
     }
+
     public long queryTimeouts()
     {
         return queryTimeouts.longValue();
     }
+
     public long annGraphSearchLatency()
     {
         return annGraphSearchLatency.longValue();
@@ -222,19 +307,6 @@ public void checkpoint()
         }
     }
 
-    public void addShadowed(long count)
-    {
-        shadowedPrimaryKeyCount.add(count);
-    }
-
-    /**
-     * @return shadowed primary keys, in ascending order
-     */
-    public long getShadowedPrimaryKeyCount()
-    {
-        return shadowedPrimaryKeyCount.longValue();
-    }
-
     public float getAnnRerankFloor()
     {
         return annRerankFloor;
@@ -277,9 +349,13 @@ public static class Snapshot
         public final long totalQueryTimeNs;
         public final long sstablesHit;
         public final long segmentsHit;
-        public final long partitionsRead;
-        public final long rowsFiltered;
-        public final long rowsPreFiltered;
+        public final long keysFetched;
+        public final long partitionsFetched;
+        public final long partitionsReturned;
+        public final long partitionTombstonesFetched;
+        public final long rowsFetched;
+        public final long rowsReturned;
+        public final long rowTombstonesFetched;
         public final long trieSegmentsHit;
         public final long bkdPostingListsHit;
         public final long bkdSegmentsHit;
@@ -289,7 +365,6 @@ public static class Snapshot
         public final long triePostingsDecodes;
         public final long queryTimeouts;
         public final long annGraphSearchLatency;
-        public final long shadowedPrimaryKeyCount;
         public final FilterSortOrder filterSortOrder;
 
         /**
@@ -302,9 +377,13 @@ private Snapshot(QueryContext context)
             totalQueryTimeNs = context.totalQueryTimeNs();
             sstablesHit = context.sstablesHit();
             segmentsHit = context.segmentsHit();
-            partitionsRead = context.partitionsRead();
-            rowsFiltered = context.rowsFiltered();
-            rowsPreFiltered = context.rowsPreFiltered();
+            keysFetched = context.keysFetched();
+            partitionsFetched = context.partitionsFetched();
+            partitionsReturned = context.partitionsReturned();
+            partitionTombstonesFetched = context.partitionTombstonesFetched();
+            rowsFetched = context.rowsFetched();
+            rowsReturned = context.rowsReturned();
+            rowTombstonesFetched = context.rowTombstonesFetched();
             trieSegmentsHit = context.trieSegmentsHit();
             bkdPostingListsHit = context.bkdPostingListsHit();
             bkdSegmentsHit = context.bkdSegmentsHit();
@@ -314,7 +393,6 @@ private Snapshot(QueryContext context)
             triePostingsDecodes = context.triePostingsDecodes();
             queryTimeouts = context.queryTimeouts();
             annGraphSearchLatency = context.annGraphSearchLatency();
-            shadowedPrimaryKeyCount = context.getShadowedPrimaryKeyCount();
             filterSortOrder = context.filterSortOrder();
         }
     }

src/java/org/apache/cassandra/index/sai/SSTableContext.java

@@ -65,7 +65,6 @@ private SSTableContext(SSTableContext copy)
         this.primaryKeyMapFactory = copy.primaryKeyMapFactory;
     }
 
-    @SuppressWarnings("resource")
     public static SSTableContext create(SSTableReader sstable, IndexComponents.ForRead perSSTableComponents)
     {
         var onDiskFormat = perSSTableComponents.onDiskFormat();

src/java/org/apache/cassandra/index/sai/StorageAttachedIndexGroup.java

@@ -246,7 +246,7 @@ public void rangeTombstone(RangeTombstone tombstone)
 
             private void forEach(Consumer<Index.Indexer> action)
             {
-                indexers.forEach(action::accept);
+                indexers.forEach(action);
             }
         };
     }
@@ -336,7 +336,7 @@ public void handleNotification(INotification notification, Object sender)
 
             // Avoid validation for index files just written following Memtable flush. ZCS streaming should
             // validate index checksum.
-            boolean validate = notice.fromStreaming() || !notice.memtable().isPresent();
+            boolean validate = notice.fromStreaming() || notice.memtable().isEmpty();
             onSSTableChanged(Collections.emptySet(), notice.added, indices, validate);
         }
         else if (notification instanceof SSTableListChangedNotification)
@@ -436,7 +436,7 @@ public int totalIndexBuildsInProgress()
      */
     public int totalQueryableIndexCount()
     {
-        return (int) indices.stream().filter(i -> baseCfs.indexManager.isIndexQueryable(i)).count();
+        return (int) indices.stream().filter(baseCfs.indexManager::isIndexQueryable).count();
     }
 
     /**
@@ -513,7 +513,7 @@ public void unsafeReload()
     public void reset()
     {
         contextManager.clear();
-        indices.forEach(index -> index.makeIndexNonQueryable());
+        indices.forEach(StorageAttachedIndex::makeIndexNonQueryable);
         onSSTableChanged(baseCfs.getLiveSSTables(), Collections.emptySet(), indices, false);
     }
 }

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

@@ -21,16 +21,14 @@
 
 import com.google.common.base.Stopwatch;
 
-import org.apache.cassandra.db.DecoratedKey;
-import org.apache.cassandra.db.rows.Unfiltered;
 import org.apache.cassandra.index.sai.utils.PrimaryKey;
 
 /**
  * Writes all SSTable-attached index token and offset structures.
  */
 public interface PerSSTableWriter
 {
-    public static final PerSSTableWriter NONE = (key) -> {};
+    PerSSTableWriter NONE = key -> {};
 
     default void startPartition(long position) throws IOException
     {}