natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Log queries scanning too many SSTables per read 

patch by Alan Wang; reviewed by David Capwell, Marcus Eriksson for CASSANDRA-21048
This commit is contained in:
Alan Wang 2025-12-12 12:38:28 -08:00 committed by David Capwell
parent a84abe7ba1
commit 25390eb8d5
9 changed files with 169 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

4
conf/cassandra.yaml
View File

@ -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.

1
src/java/org/apache/cassandra/config/Config.java
View File

@ -579,6 +579,7 @@ public class Config
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;

10
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -3215,6 +3215,16 @@ public class DatabaseDescriptor
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;

5
src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
View File

@ -58,12 +58,14 @@ import org.apache.cassandra.service.reads.ReadCoordinator;
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 class PartitionRangeReadCommand extends ReadCommand implements PartitionR
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());

13
src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
View File

@ -87,12 +87,14 @@ import org.apache.cassandra.tracing.Tracing;
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 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
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 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
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);

11
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -4719,6 +4719,17 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
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();

5
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -1036,6 +1036,11 @@ public interface StorageServiceMBean extends NotificationEmitter
/** 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 */

120
test/distributed/org/apache/cassandra/distributed/test/SSTableReadLogsQueryTest.java Normal file
View File

@ -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());
}
}
}