Log queries scanning too many SSTables per read

patch by Alan Wang; reviewed by David Capwell, Marcus Eriksson for CASSANDRA-21048

Author: alanwang67

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Log queries scanning too many SSTables per read (CASSANDRA-21048)
  * Extend nodetool verify to (optionally) validate SAI files (CASSANDRA-20949)
  * Fix CompressionDictionary being closed while still in use (CASSANDRA-21047)
  * When updating a multi cell collection element, if the update is rejected then the shared Row.Builder is not freed causing all future mutations to be rejected (CASSANDRA-21055)

conf/cassandra.yaml

@@ -1990,6 +1990,10 @@ transparent_data_encryption_options:
 # SAFETY THRESHOLDS #
 #####################
 
+# Log the query if it reads more than
+# sstables_per_read_log_threshold SSTables
+sstables_per_read_log_threshold: 100
+
 # When executing a scan, within or across a partition, we need to keep the
 # tombstones seen in memory so we can return them to the coordinator, which
 # will use them to make sure other replicas also know about the deleted rows.

src/java/org/apache/cassandra/config/Config.java

@@ -579,6 +579,7 @@ public static class SSTableConfig
     public volatile DataStorageSpec.LongBytesBound row_index_read_size_warn_threshold = null;
     public volatile DataStorageSpec.LongBytesBound row_index_read_size_fail_threshold = null;
 
+    public volatile int sstables_per_read_log_threshold = 100;
     public volatile int tombstone_warn_threshold = 1000;
     public volatile int tombstone_failure_threshold = 100000;
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -3215,6 +3215,16 @@ public static int getMaxMutationSize()
         return conf.max_mutation_size.toBytes();
     }
 
+    public static int getSSTablesPerReadLogThreshold()
+    {
+        return conf.sstables_per_read_log_threshold;
+    }
+
+    public static void setSSTablesPerReadLogThreshold(int threshold)
+    {
+        conf.sstables_per_read_log_threshold = threshold;
+    }
+
     public static int getTombstoneWarnThreshold()
     {
         return conf.tombstone_warn_threshold;

src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java

@@ -58,12 +58,14 @@
 import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.transport.Dispatcher;
+import org.apache.cassandra.utils.NoSpamLogger;
 
 /**
  * A read command that selects a (part of a) range of partitions.
  */
 public class PartitionRangeReadCommand extends ReadCommand implements PartitionRangeReadQuery
 {
+    private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.SECONDS);
     protected static final SelectionDeserializer selectionDeserializer = new Deserializer();
 
     protected final Slices requestedSlices;
@@ -434,6 +436,9 @@ public UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, Rea
 
             final int finalSelectedSSTables = selectedSSTablesCnt;
 
+            if (finalSelectedSSTables > DatabaseDescriptor.getSSTablesPerReadLogThreshold())
+                noSpamLogger.info("The following query '{}' has read {} SSTables.", this.toCQLString(), finalSelectedSSTables);
+
             // iterators can be empty for offline tools
             if (inputCollector.isEmpty())
                 return EmptyIterators.unfilteredPartition(metadata());

src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java

@@ -87,12 +87,14 @@
 import org.apache.cassandra.transport.Dispatcher;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.btree.BTreeSet;
+import org.apache.cassandra.utils.NoSpamLogger;
 
 /**
  * A read command that selects a (part of a) single partition.
  */
 public class SinglePartitionReadCommand extends ReadCommand implements SinglePartitionReadQuery
 {
+    private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.SECONDS);
     protected static final SelectionDeserializer selectionDeserializer = new Deserializer();
     protected static final Function<Seekable, SelectionDeserializer> accordSelectionDeserializer = AccordDeserializer::new;
 
@@ -876,7 +878,13 @@ private UnfilteredRowIterator queryMemtableAndDiskInternal(ColumnFamilyStore cfs
             StorageHook.instance.reportRead(cfs.metadata().id, partitionKey());
 
             List<UnfilteredRowIterator> iterators = inputCollector.finalizeIterators(cfs, nowInSec(), controller.oldestUnrepairedTombstone());
-            return withSSTablesIterated(iterators, cfs.metric, metricsCollector);
+
+            UnfilteredRowIterator result = withSSTablesIterated(iterators, cfs.metric, metricsCollector);
+
+            if (metricsCollector.getMergedSSTables() > DatabaseDescriptor.getSSTablesPerReadLogThreshold())
+                noSpamLogger.info("The following query '{}' has read {} SSTables.", this.toCQLString(), metricsCollector.getMergedSSTables());
+
+            return result;
         }
         catch (RuntimeException | Error e)
         {
@@ -1084,6 +1092,9 @@ private UnfilteredRowIterator queryMemtableAndSSTablesInTimestampOrder(ColumnFam
 
         cfs.metric.updateSSTableIterated(metricsCollector.getMergedSSTables());
 
+        if (metricsCollector.getMergedSSTables() > DatabaseDescriptor.getSSTablesPerReadLogThreshold())
+            noSpamLogger.info("The following query '{}' has read {} SSTables.", this.toCQLString(), metricsCollector.getMergedSSTables());
+
         if (result == null || result.isEmpty())
             return EmptyIterators.unfilteredRow(metadata(), partitionKey(), false);
 

src/java/org/apache/cassandra/service/StorageService.java

@@ -4719,6 +4719,17 @@ public void setInvalidateKeycacheOnSSTableDeletion(boolean invalidate)
         DatabaseDescriptor.setInvalidateKeycacheOnSSTableDeletion(invalidate);
     }
 
+    public int getSSTablesPerReadLogThreshold()
+    {
+        return DatabaseDescriptor.getSSTablesPerReadLogThreshold();
+    }
+
+    public void setSSTablesPerReadLogThreshold(int threshold)
+    {
+        DatabaseDescriptor.setSSTablesPerReadLogThreshold(threshold);
+        logger.info("updated sstables_per_read_log_threshold to {}", threshold);
+    }
+
     public int getTombstoneWarnThreshold()
     {
         return DatabaseDescriptor.getTombstoneWarnThreshold();

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -1036,6 +1036,11 @@ default int upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion,
     /** Returns the cluster partitioner */
     public String getPartitionerName();
 
+    /** Returns the threshold for logging queries that read more than threshold amount of SSTables */
+    public int getSSTablesPerReadLogThreshold();
+    /** Sets the threshold for logging queries that read more than threshold amount of SSTables */
+    public void setSSTablesPerReadLogThreshold(int threshold);
+
     /** Returns the threshold for warning of queries with many tombstones */
     public int getTombstoneWarnThreshold();
     /** Sets the threshold for warning queries with many tombstones */

test/distributed/org/apache/cassandra/distributed/test/SSTableReadLogsQueryTest.java

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.Cluster;
+
+public class SSTableReadLogsQueryTest extends TestBaseImpl
+{
+    @Test
+    public void logQueryTest() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.build(1)
+                                           .start()))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int PRIMARY KEY, v counter)");
+
+            cluster.get(1).runOnInstance(() -> {
+                Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").disableAutoCompaction();
+            });
+
+            for (int i = 0; i <= 100; i++)
+            {
+                cluster.get(1).executeInternal("UPDATE " + KEYSPACE + ".tbl SET v = v + 1 WHERE pk = 2");
+                cluster.get(1).flush(withKeyspace("%s"));
+            }
+
+            cluster.get(1).runOnInstance(() -> {
+                assertEquals(101, Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").getTracker().getView().liveSSTables().size());
+            });
+
+            String query = "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 2";
+            cluster.get(1).executeInternalWithResult(query);
+
+            assertEquals(1, cluster.get(1).logs().watchFor("The following query").getResult().size());
+        }
+    }
+
+    @Test
+    public void setSSTablesPerReadLogThresholdTest() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.build(1)
+                                           .start()))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int PRIMARY KEY, v counter)");
+
+            cluster.get(1).runOnInstance(() -> {
+                Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").disableAutoCompaction();
+            });
+
+            cluster.get(1).runOnInstance(() -> {
+                DatabaseDescriptor.setSSTablesPerReadLogThreshold(25);
+            });
+
+            for (int i = 0; i <= 25; i++)
+            {
+                cluster.get(1).executeInternal("UPDATE " + KEYSPACE + ".tbl SET v = v + 1 WHERE pk = 2");
+                cluster.get(1).flush(withKeyspace("%s"));
+            }
+
+            cluster.get(1).runOnInstance(() -> {
+                assertEquals(26, Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").getTracker().getView().liveSSTables().size());
+            });
+
+            String query = "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 2";
+            cluster.get(1).executeInternalWithResult(query);
+
+            assertEquals(1, cluster.get(1).logs().watchFor("The following query").getResult().size());
+        }
+    }
+
+    @Test
+    public void logRangeReadQueryTest() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.build(1)
+                                           .start()))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int PRIMARY KEY, v int)");
+
+            cluster.get(1).runOnInstance(() -> {
+                Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").disableAutoCompaction();
+            });
+
+            for (int i = 0; i <= 100; i++)
+            {
+                cluster.get(1).executeInternal(String.format("INSERT INTO " + KEYSPACE + ".tbl (pk, v) VALUES (%s, %s)", i, i));
+                cluster.get(1).flush(withKeyspace("%s"));
+            }
+
+            cluster.get(1).runOnInstance(() -> {
+                assertEquals(101, Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").getTracker().getView().liveSSTables().size());
+            });
+
+            String query = "SELECT * FROM " + KEYSPACE + ".tbl WHERE pk >= 0 AND pk < 51 ALLOW FILTERING";
+            cluster.get(1).executeInternalWithResult(query);
+
+            assertEquals(1, cluster.get(1).logs().watchFor("The following query").getResult().size());
+        }
+    }
+}