diff --git a/CHANGES.txt b/CHANGES.txt
index 2501253fac..388fad1519 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.1
+ * Add soft/hard limits to local reads to protect against reading too much data in a single query (CASSANDRA-16896)
* Avoid token cache invalidation for removing a non-member node (CASSANDRA-15290)
* Allow configuration of consistency levels on auth operations (CASSANDRA-12988)
* Add number of sstables in a compaction to compactionstats output (CASSANDRA-16844)
diff --git a/NEWS.txt b/NEWS.txt
index 77a3e1bf72..bdd3dd6ffe 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -40,11 +40,14 @@ New features
------------
- Warn/abort thresholds added to read queries notifying clients when these thresholds trigger (by
emitting a client warning or aborting the query). This feature is disabled by default, scheduled
- to be enabled in 4.2; it is controlled with the configuration client_track_warnings_enabled,
+ to be enabled in 4.2; it is controlled with the configuration track_warnings.enabled,
setting to true will enable this feature. Each check has its own warn/abort thresholds, currently
- tombstones (tombstone_warn_threshold, and tombstone_failure_threshold) and coordinator result set
- materialized size (client_large_read_warn_threshold_kb, and client_large_read_abort_threshold_kb)
- are supported; more checks will be added over time.
+ tombstones (tombstone_warn_threshold, and tombstone_failure_threshold), coordinator result set
+ materialized size (track_warnings.coordinator_large_read.warn_threshold_kb, and
+ track_warnings.coordinator_large_read.abort_threshold_kb), local read materialized heap size
+ (track_warnings.local_read_size.warn_threshold_kb and track_warnings.local_read_size.abort_threshold_kb),
+ and RowIndexEntry estimated memory size (track_warnings.row_index_size.warn_threshold_kb and
+ track_warnings.row_index_size.abort_threshold_kb) are supported; more checks will be added over time.
Upgrading
---------
diff --git a/build.xml b/build.xml
index 32eba7c26a..1dc1c035cd 100644
--- a/build.xml
+++ b/build.xml
@@ -859,6 +859,7 @@
+
@@ -1370,6 +1371,7 @@
+
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index a868a4af0a..4e6db631be 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -1459,15 +1459,25 @@ enable_drop_compact_storage: false
# - 127.0.0.0/31
# Enables tracking warnings/aborts across all replicas for reporting back to client.
-# Scheduled to enable in 4.2
# See: CASSANDRA-16850
-# See: tombstone_warn_threshold, tombstone_failure_threshold, client_large_read_warn_threshold_kb, and client_large_read_abort_threshold_kb
-#client_track_warnings_enabled: false
-
-# When client_track_warnings_enabled: true, this tracks the materialized size of a query on the
-# coordinator. If client_large_read_warn_threshold_kb is greater than 0, this will emit a warning
-# to clients with details on what query triggered this as well as the size of the result set; if
-# client_large_read_abort_threshold_kb is greater than 0, this will abort the query after it
-# has exceeded this threshold, returning a read error to the user.
-#client_large_read_warn_threshold_kb: 0
-#client_large_read_abort_threshold_kb: 0
+#track_warnings:
+# # Scheduled to enable in 4.2
+# enabled: false
+# # When track_warnings.enabled: true, this tracks the materialized size of a query on the
+# # coordinator. If coordinator_large_read.warn_threshold_kb is greater than 0, this will emit a warning
+# # to clients with details on what query triggered this as well as the size of the result set; if
+# # coordinator_large_read.abort_threshold_kb is greater than 0, this will abort the query after it
+# # has exceeded this threshold, returning a read error to the user.
+# coordinator_large_read:
+# warn_threshold_kb: 0
+# abort_threshold_kb: 0
+# # When track_warnings.enabled: true, this tracks the size of the local read (as defined by
+# # heap size), and will warn/abort based off these thresholds; 0 disables these checks.
+# local_read_size:
+# warn_threshold_kb: 0
+# abort_threshold_kb: 0
+# # When track_warnings.enabled: true, this tracks the expected memory size of the RowIndexEntry
+# # and will warn/abort based off these thresholds; 0 disables these checks.
+# row_index_size:
+# warn_threshold_kb: 0
+# abort_threshold_kb: 0
diff --git a/ide/idea/workspace.xml b/ide/idea/workspace.xml
index 41645f5ce9..73af47fcc8 100644
--- a/ide/idea/workspace.xml
+++ b/ide/idea/workspace.xml
@@ -143,7 +143,7 @@
-
+
@@ -167,7 +167,7 @@
-
+
@@ -186,7 +186,7 @@
-
+
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index 4d4cc11a76..caa496536d 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -346,19 +346,16 @@ public class Config
public MemtableAllocationType memtable_allocation_type = MemtableAllocationType.heap_buffers;
+ public final TrackWarnings track_warnings = new TrackWarnings();
+
public volatile int tombstone_warn_threshold = 1000;
public volatile int tombstone_failure_threshold = 100000;
- public volatile long client_large_read_warn_threshold_kb = 0;
- public volatile long client_large_read_abort_threshold_kb = 0;
-
public final ReplicaFilteringProtectionOptions replica_filtering_protection = new ReplicaFilteringProtectionOptions();
public volatile Long index_summary_capacity_in_mb;
public volatile int index_summary_resize_interval_in_minutes = 60;
- public volatile boolean client_track_warnings_enabled = false; // should set to true in 4.2
-
public int gc_log_threshold_in_ms = 200;
public int gc_warn_threshold_in_ms = 1000;
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 511ef3f0d1..32a333b0b5 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -688,6 +688,7 @@ public class DatabaseDescriptor
applyConcurrentValidations(conf);
applyRepairCommandPoolSize(conf);
+ applyTrackWarningsValidations(conf);
if (conf.concurrent_materialized_view_builders <= 0)
throw new ConfigurationException("concurrent_materialized_view_builders should be strictly greater than 0, but was " + conf.concurrent_materialized_view_builders, false);
@@ -857,9 +858,6 @@ public class DatabaseDescriptor
throw new ConfigurationException("To set concurrent_validations > concurrent_compactors, " +
"set the system property cassandra.allow_unlimited_concurrent_validations=true");
}
-
- conf.client_large_read_warn_threshold_kb = Math.max(conf.client_large_read_warn_threshold_kb, 0);
- conf.client_large_read_abort_threshold_kb = Math.max(conf.client_large_read_abort_threshold_kb, 0);
}
@VisibleForTesting
@@ -869,6 +867,12 @@ public class DatabaseDescriptor
config.repair_command_pool_size = config.concurrent_validations;
}
+ @VisibleForTesting
+ static void applyTrackWarningsValidations(Config config)
+ {
+ config.track_warnings.validate("track_warnings");
+ }
+
private static String storagedirFor(String type)
{
return storagedir(type + "_directory") + File.separator + type;
@@ -3477,33 +3481,73 @@ public class DatabaseDescriptor
return conf.internode_error_reporting_exclusions;
}
- public static long getClientLargeReadWarnThresholdKB()
+ public static boolean getTrackWarningsEnabled()
{
- return conf.client_large_read_warn_threshold_kb;
+ return conf.track_warnings.enabled;
}
- public static void setClientLargeReadWarnThresholdKB(long threshold)
+ public static void setTrackWarningsEnabled(boolean value)
{
- conf.client_large_read_warn_threshold_kb = Math.max(threshold, 0);
+ conf.track_warnings.enabled = value;
}
- public static long getClientLargeReadAbortThresholdKB()
+ public static long getCoordinatorReadSizeWarnThresholdKB()
{
- return conf.client_large_read_abort_threshold_kb;
+ return conf.track_warnings.coordinator_read_size.getWarnThresholdKb();
}
- public static void setClientLargeReadAbortThresholdKB(long threshold)
+ public static void setCoordinatorReadSizeWarnThresholdKB(long threshold)
{
- conf.client_large_read_abort_threshold_kb = Math.max(threshold, 0);
+ conf.track_warnings.coordinator_read_size.setWarnThresholdKb(threshold);
}
- public static boolean getClientTrackWarningsEnabled()
+ public static long getCoordinatorReadSizeAbortThresholdKB()
{
- return conf.client_track_warnings_enabled;
+ return conf.track_warnings.coordinator_read_size.getAbortThresholdKb();
}
- public static void setClientTrackWarningsEnabled(boolean value)
+ public static void setCoordinatorReadSizeAbortThresholdKB(long threshold)
{
- conf.client_track_warnings_enabled = value;
+ conf.track_warnings.coordinator_read_size.setAbortThresholdKb(threshold);
+ }
+
+ public static long getLocalReadSizeWarnThresholdKb()
+ {
+ return conf.track_warnings.local_read_size.getWarnThresholdKb();
+ }
+
+ public static void setLocalReadSizeWarnThresholdKb(long value)
+ {
+ conf.track_warnings.local_read_size.setWarnThresholdKb(value);
+ }
+
+ public static long getLocalReadSizeAbortThresholdKb()
+ {
+ return conf.track_warnings.local_read_size.getAbortThresholdKb();
+ }
+
+ public static void setLocalReadSizeAbortThresholdKb(long value)
+ {
+ conf.track_warnings.local_read_size.setAbortThresholdKb(value);
+ }
+
+ public static int getRowIndexSizeWarnThresholdKb()
+ {
+ return conf.track_warnings.row_index_size.getWarnThresholdKb();
+ }
+
+ public static void setRowIndexSizeWarnThresholdKb(int value)
+ {
+ conf.track_warnings.row_index_size.setWarnThresholdKb(value);
+ }
+
+ public static int getRowIndexSizeAbortThresholdKb()
+ {
+ return conf.track_warnings.row_index_size.getAbortThresholdKb();
+ }
+
+ public static void setRowIndexSizeAbortThresholdKb(int value)
+ {
+ conf.track_warnings.row_index_size.setAbortThresholdKb(value);
}
}
diff --git a/src/java/org/apache/cassandra/config/TrackWarnings.java b/src/java/org/apache/cassandra/config/TrackWarnings.java
new file mode 100644
index 0000000000..77530a8e8b
--- /dev/null
+++ b/src/java/org/apache/cassandra/config/TrackWarnings.java
@@ -0,0 +1,108 @@
+/*
+ * 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.config;
+
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+public class TrackWarnings
+{
+ public volatile boolean enabled = false; // should set to true in 4.2
+ public final LongByteThreshold coordinator_read_size = new LongByteThreshold();
+ public final LongByteThreshold local_read_size = new LongByteThreshold();
+ public final IntByteThreshold row_index_size = new IntByteThreshold();
+
+ public void validate(String prefix)
+ {
+ prefix += ".";
+ coordinator_read_size.validate(prefix + "coordinator_read_size");
+ local_read_size.validate(prefix + "local_read_size");
+ row_index_size.validate(prefix + "row_index_size");
+ }
+
+ public static class LongByteThreshold
+ {
+ public volatile long warn_threshold_kb = 0;
+ public volatile long abort_threshold_kb = 0;
+
+ public long getWarnThresholdKb()
+ {
+ return warn_threshold_kb;
+ }
+
+ public void setWarnThresholdKb(long value)
+ {
+ warn_threshold_kb = Math.max(value, 0);
+ }
+
+ public long getAbortThresholdKb()
+ {
+ return abort_threshold_kb;
+ }
+
+ public void setAbortThresholdKb(long value)
+ {
+ abort_threshold_kb = Math.max(value, 0);
+ }
+
+ public void validate(String prefix)
+ {
+ warn_threshold_kb = Math.max(warn_threshold_kb, 0);
+ abort_threshold_kb = Math.max(abort_threshold_kb, 0);
+
+ if (abort_threshold_kb != 0 && abort_threshold_kb < warn_threshold_kb)
+ throw new ConfigurationException(String.format("abort_threshold_kb (%d) must be greater than or equal to warn_threshold_kb (%d); see %s",
+ abort_threshold_kb, warn_threshold_kb, prefix));
+ }
+ }
+
+ public static class IntByteThreshold
+ {
+ public volatile int warn_threshold_kb = 0;
+ public volatile int abort_threshold_kb = 0;
+
+ public int getWarnThresholdKb()
+ {
+ return warn_threshold_kb;
+ }
+
+ public void setWarnThresholdKb(int value)
+ {
+ warn_threshold_kb = Math.max(value, 0);
+ }
+
+ public int getAbortThresholdKb()
+ {
+ return abort_threshold_kb;
+ }
+
+ public void setAbortThresholdKb(int value)
+ {
+ abort_threshold_kb = Math.max(value, 0);
+ }
+
+ public void validate(String prefix)
+ {
+ warn_threshold_kb = Math.max(warn_threshold_kb, 0);
+ abort_threshold_kb = Math.max(abort_threshold_kb, 0);
+
+ if (abort_threshold_kb != 0 && abort_threshold_kb < warn_threshold_kb)
+ throw new ConfigurationException(String.format("abort_threshold_kb (%d) must be greater than or equal to warn_threshold_kb (%d); see %s",
+ abort_threshold_kb, warn_threshold_kb, prefix));
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/cql3/QueryOptions.java b/src/java/org/apache/cassandra/cql3/QueryOptions.java
index e46c45897e..7e3d267ed7 100644
--- a/src/java/org/apache/cassandra/cql3/QueryOptions.java
+++ b/src/java/org/apache/cassandra/cql3/QueryOptions.java
@@ -219,19 +219,19 @@ public abstract class QueryOptions
abstract TrackWarnings getTrackWarnings();
- public boolean isClientTrackWarningsEnabled()
+ public boolean isTrackWarningsEnabled()
{
return getTrackWarnings().isEnabled();
}
- public long getClientLargeReadWarnThresholdKb()
+ public long getCoordinatorReadSizeWarnThresholdKB()
{
- return getTrackWarnings().getClientLargeReadWarnThresholdKb();
+ return getTrackWarnings().getCoordinatorReadSizeWarnThresholdKB();
}
- public long getClientLargeReadAbortThresholdKB()
+ public long getCoordinatorReadSizeAbortThresholdKB()
{
- return getTrackWarnings().getClientLargeReadAbortThresholdKB();
+ return getTrackWarnings().getCoordinatorReadSizeAbortThresholdKB();
}
public QueryOptions prepare(List specs)
@@ -243,21 +243,21 @@ public abstract class QueryOptions
{
boolean isEnabled();
- long getClientLargeReadWarnThresholdKb();
+ long getCoordinatorReadSizeWarnThresholdKB();
- long getClientLargeReadAbortThresholdKB();
+ long getCoordinatorReadSizeAbortThresholdKB();
static TrackWarnings create()
{
// if daemon initialization hasn't happened yet (very common in tests) then ignore
if (!DatabaseDescriptor.isDaemonInitialized())
return DisabledTrackWarnings.INSTANCE;
- boolean enabled = DatabaseDescriptor.getClientTrackWarningsEnabled();
+ boolean enabled = DatabaseDescriptor.getTrackWarningsEnabled();
if (!enabled)
return DisabledTrackWarnings.INSTANCE;
- long clientLargeReadWarnThresholdKb = DatabaseDescriptor.getClientLargeReadWarnThresholdKB();
- long clientLargeReadAbortThresholdKB = DatabaseDescriptor.getClientLargeReadAbortThresholdKB();
- return new DefaultTrackWarnings(clientLargeReadWarnThresholdKb, clientLargeReadAbortThresholdKB);
+ long warnThresholdKB = DatabaseDescriptor.getCoordinatorReadSizeWarnThresholdKB();
+ long abortThresholdKB = DatabaseDescriptor.getCoordinatorReadSizeAbortThresholdKB();
+ return new DefaultTrackWarnings(warnThresholdKB, abortThresholdKB);
}
}
@@ -272,13 +272,13 @@ public abstract class QueryOptions
}
@Override
- public long getClientLargeReadWarnThresholdKb()
+ public long getCoordinatorReadSizeWarnThresholdKB()
{
return 0;
}
@Override
- public long getClientLargeReadAbortThresholdKB()
+ public long getCoordinatorReadSizeAbortThresholdKB()
{
return 0;
}
@@ -286,13 +286,13 @@ public abstract class QueryOptions
private static class DefaultTrackWarnings implements TrackWarnings
{
- private final long clientLargeReadWarnThresholdKb;
- private final long clientLargeReadAbortThresholdKB;
+ private final long warnThresholdKB;
+ private final long abortThresholdKB;
- public DefaultTrackWarnings(long clientLargeReadWarnThresholdKb, long clientLargeReadAbortThresholdKB)
+ public DefaultTrackWarnings(long warnThresholdKB, long abortThresholdKB)
{
- this.clientLargeReadWarnThresholdKb = clientLargeReadWarnThresholdKb;
- this.clientLargeReadAbortThresholdKB = clientLargeReadAbortThresholdKB;
+ this.warnThresholdKB = warnThresholdKB;
+ this.abortThresholdKB = abortThresholdKB;
}
@Override
@@ -302,15 +302,15 @@ public abstract class QueryOptions
}
@Override
- public long getClientLargeReadWarnThresholdKb()
+ public long getCoordinatorReadSizeWarnThresholdKB()
{
- return clientLargeReadWarnThresholdKb;
+ return warnThresholdKB;
}
@Override
- public long getClientLargeReadAbortThresholdKB()
+ public long getCoordinatorReadSizeAbortThresholdKB()
{
- return clientLargeReadAbortThresholdKB;
+ return abortThresholdKB;
}
}
diff --git a/src/java/org/apache/cassandra/cql3/selection/ResultSetBuilder.java b/src/java/org/apache/cassandra/cql3/selection/ResultSetBuilder.java
index 852872a778..df82d527d4 100644
--- a/src/java/org/apache/cassandra/cql3/selection/ResultSetBuilder.java
+++ b/src/java/org/apache/cassandra/cql3/selection/ResultSetBuilder.java
@@ -106,6 +106,11 @@ public final class ResultSetBuilder
return thresholdKB > 0 && size > thresholdKB << 10;
}
+ public long getSize()
+ {
+ return size;
+ }
+
public void add(ByteBuffer v)
{
current.add(v);
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 5c7ac2952f..25499b2171 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -31,7 +31,6 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.audit.AuditLogContext;
import org.apache.cassandra.audit.AuditLogEntryType;
import org.apache.cassandra.auth.Permission;
-import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
@@ -248,7 +247,7 @@ public class SelectStatement implements CQLStatement
Selectors selectors = selection.newSelectors(options);
ReadQuery query = getQuery(options, selectors.getColumnFilter(), nowInSec, userLimit, userPerPartitionLimit, pageSize);
- if (options.isClientTrackWarningsEnabled())
+ if (options.isTrackWarningsEnabled())
query.trackWarnings();
if (aggregationSpec == null && (pageSize <= 0 || (query.limits().count() <= pageSize)))
@@ -816,32 +815,41 @@ public class SelectStatement implements CQLStatement
private void maybeWarn(ResultSetBuilder result, QueryOptions options)
{
- if (!options.isClientTrackWarningsEnabled())
+ if (!options.isTrackWarningsEnabled())
return;
- if (result.shouldWarn(options.getClientLargeReadWarnThresholdKb()))
+ ColumnFamilyStore store = cfs();
+ if (store != null)
+ store.metric.coordinatorReadSize.update(result.getSize());
+ if (result.shouldWarn(options.getCoordinatorReadSizeWarnThresholdKB()))
{
- String msg = String.format("Read on table %s has exceeded the size warning threshold of %,d kb", table, options.getClientLargeReadWarnThresholdKb());
+ String msg = String.format("Read on table %s has exceeded the size warning threshold of %,d kb", table, options.getCoordinatorReadSizeWarnThresholdKB());
ClientWarn.instance.warn(msg + " with " + loggableTokens(options));
logger.warn("{} with query {}", msg, asCQL(options));
- cfs().metric.clientReadSizeWarnings.mark();
+ if (store != null)
+ store.metric.coordinatorReadSizeWarnings.mark();
}
}
private void maybeFail(ResultSetBuilder result, QueryOptions options)
{
- if (!options.isClientTrackWarningsEnabled())
+ if (!options.isTrackWarningsEnabled())
return;
- if (result.shouldReject(options.getClientLargeReadAbortThresholdKB()))
+ if (result.shouldReject(options.getCoordinatorReadSizeAbortThresholdKB()))
{
- String msg = String.format("Read on table %s has exceeded the size failure threshold of %,d kb", table, options.getClientLargeReadAbortThresholdKB());
+ String msg = String.format("Read on table %s has exceeded the size failure threshold of %,d kb", table, options.getCoordinatorReadSizeAbortThresholdKB());
String clientMsg = msg + " with " + loggableTokens(options);
ClientWarn.instance.warn(clientMsg);
logger.warn("{} with query {}", msg, asCQL(options));
- cfs().metric.clientReadSizeAborts.mark();
+ ColumnFamilyStore store = cfs();
+ if (store != null)
+ {
+ store.metric.coordinatorReadSizeAborts.mark();
+ store.metric.coordinatorReadSize.update(result.getSize());
+ }
// read errors require blockFor and recieved (its in the protocol message), but this isn't known;
// to work around this, treat the coordinator as the only response we care about and mark it failed
ReadSizeAbortException exception = new ReadSizeAbortException(clientMsg, options.getConsistency(), 0, 1, true,
- ImmutableMap.of(FBUtilities.getBroadcastAddressAndPort(), RequestFailureReason.READ_TOO_LARGE));
+ ImmutableMap.of(FBUtilities.getBroadcastAddressAndPort(), RequestFailureReason.READ_SIZE));
StorageProxy.recordReadRegularAbort(options.getConsistency(), exception);
throw exception;
}
diff --git a/src/java/org/apache/cassandra/db/ArrayClustering.java b/src/java/org/apache/cassandra/db/ArrayClustering.java
index a6ee991278..53d45e7474 100644
--- a/src/java/org/apache/cassandra/db/ArrayClustering.java
+++ b/src/java/org/apache/cassandra/db/ArrayClustering.java
@@ -22,7 +22,7 @@ import org.apache.cassandra.utils.ObjectSizes;
public class ArrayClustering extends AbstractArrayClusteringPrefix implements Clustering
{
- private static final long EMPTY_SIZE = ObjectSizes.measure(new ArrayClustering(EMPTY_VALUES_ARRAY));
+ public static final long EMPTY_SIZE = ObjectSizes.measure(new ArrayClustering(EMPTY_VALUES_ARRAY));
public ArrayClustering(byte[]... values)
{
diff --git a/src/java/org/apache/cassandra/db/Clustering.java b/src/java/org/apache/cassandra/db/Clustering.java
index f5184e99f5..7575c14fd2 100644
--- a/src/java/org/apache/cassandra/db/Clustering.java
+++ b/src/java/org/apache/cassandra/db/Clustering.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
+import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.marshal.ByteArrayAccessor;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
@@ -34,7 +35,7 @@ import org.apache.cassandra.utils.memory.AbstractAllocator;
import static org.apache.cassandra.db.AbstractBufferClusteringPrefix.EMPTY_VALUES_ARRAY;
-public interface Clustering extends ClusteringPrefix
+public interface Clustering extends ClusteringPrefix, IMeasurableMemory
{
public static final Serializer serializer = new Serializer();
diff --git a/src/java/org/apache/cassandra/db/DeletionTime.java b/src/java/org/apache/cassandra/db/DeletionTime.java
index d8ac91db98..105e10ddb4 100644
--- a/src/java/org/apache/cassandra/db/DeletionTime.java
+++ b/src/java/org/apache/cassandra/db/DeletionTime.java
@@ -33,7 +33,7 @@ import org.apache.cassandra.utils.ObjectSizes;
*/
public class DeletionTime implements Comparable, IMeasurableMemory
{
- private static final long EMPTY_SIZE = ObjectSizes.measure(new DeletionTime(0, 0));
+ public static final long EMPTY_SIZE = ObjectSizes.measure(new DeletionTime(0, 0));
/**
* A special DeletionTime that signifies that there is no top-level (row) tombstone.
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index f029bac8f7..4ea589a340 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -33,6 +33,7 @@ import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.config.*;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.db.filter.*;
@@ -67,6 +68,7 @@ import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.ObjectSizes;
import static com.google.common.collect.Iterables.any;
import static com.google.common.collect.Iterables.filter;
@@ -86,6 +88,10 @@ public abstract class ReadCommand extends AbstractReadQuery
protected static final Logger logger = LoggerFactory.getLogger(ReadCommand.class);
public static final IVersionedSerializer serializer = new Serializer();
+ // Expose the active command running so transitive calls can lookup this command.
+ // This is useful for a few reasons, but mainly because the CQL query is here.
+ private static final FastThreadLocal COMMAND = new FastThreadLocal<>();
+
private final Kind kind;
private final boolean isDigestQuery;
@@ -150,6 +156,11 @@ public abstract class ReadCommand extends AbstractReadQuery
this.trackWarnings = trackWarnings;
}
+ public static ReadCommand getCommand()
+ {
+ return COMMAND.get();
+ }
+
protected abstract void serializeSelection(DataOutputPlus out, int version) throws IOException;
protected abstract long selectionSerializedSize(int version);
@@ -388,66 +399,75 @@ public abstract class ReadCommand extends AbstractReadQuery
{
long startTimeNanos = System.nanoTime();
- ColumnFamilyStore cfs = Keyspace.openAndGetStore(metadata());
- Index index = getIndex(cfs);
-
- Index.Searcher searcher = null;
- if (index != null)
- {
- if (!cfs.indexManager.isIndexQueryable(index))
- throw new IndexNotAvailableException(index);
-
- searcher = index.searcherFor(this);
- Tracing.trace("Executing read on {}.{} using index {}", cfs.metadata.keyspace, cfs.metadata.name, index.getIndexMetadata().name);
- }
-
- UnfilteredPartitionIterator iterator = (null == searcher) ? queryStorage(cfs, executionController) : searcher.search(executionController);
- iterator = RTBoundValidator.validate(iterator, Stage.MERGED, false);
-
+ COMMAND.set(this);
try
{
- iterator = withStateTracking(iterator);
- iterator = RTBoundValidator.validate(withoutPurgeableTombstones(iterator, cfs, executionController), Stage.PURGED, false);
- iterator = withMetricsRecording(iterator, cfs.metric, startTimeNanos);
+ ColumnFamilyStore cfs = Keyspace.openAndGetStore(metadata());
+ Index index = getIndex(cfs);
- // If we've used a 2ndary index, we know the result already satisfy the primary expression used, so
- // no point in checking it again.
- RowFilter filter = (null == searcher) ? rowFilter() : index.getPostIndexQueryFilter(rowFilter());
-
- /*
- * TODO: We'll currently do filtering by the rowFilter here because it's convenient. However,
- * we'll probably want to optimize by pushing it down the layer (like for dropped columns) as it
- * would be more efficient (the sooner we discard stuff we know we don't care, the less useless
- * processing we do on it).
- */
- iterator = filter.filter(iterator, nowInSec());
-
- // apply the limits/row counter; this transformation is stopping and would close the iterator as soon
- // as the count is observed; if that happens in the middle of an open RT, its end bound will not be included.
- // If tracking repaired data, the counter is needed for overreading repaired data, otherwise we can
- // optimise the case where this.limit = DataLimits.NONE which skips an unnecessary transform
- if (executionController.isTrackingRepairedStatus())
+ Index.Searcher searcher = null;
+ if (index != null)
{
- DataLimits.Counter limit =
+ if (!cfs.indexManager.isIndexQueryable(index))
+ throw new IndexNotAvailableException(index);
+
+ searcher = index.searcherFor(this);
+ Tracing.trace("Executing read on {}.{} using index {}", cfs.metadata.keyspace, cfs.metadata.name, index.getIndexMetadata().name);
+ }
+
+ UnfilteredPartitionIterator iterator = (null == searcher) ? queryStorage(cfs, executionController) : searcher.search(executionController);
+ iterator = RTBoundValidator.validate(iterator, Stage.MERGED, false);
+
+ try
+ {
+ iterator = withQuerySizeTracking(iterator);
+ iterator = withStateTracking(iterator);
+ iterator = RTBoundValidator.validate(withoutPurgeableTombstones(iterator, cfs, executionController), Stage.PURGED, false);
+ iterator = withMetricsRecording(iterator, cfs.metric, startTimeNanos);
+
+ // If we've used a 2ndary index, we know the result already satisfy the primary expression used, so
+ // no point in checking it again.
+ RowFilter filter = (null == searcher) ? rowFilter() : index.getPostIndexQueryFilter(rowFilter());
+
+ /*
+ * TODO: We'll currently do filtering by the rowFilter here because it's convenient. However,
+ * we'll probably want to optimize by pushing it down the layer (like for dropped columns) as it
+ * would be more efficient (the sooner we discard stuff we know we don't care, the less useless
+ * processing we do on it).
+ */
+ iterator = filter.filter(iterator, nowInSec());
+
+ // apply the limits/row counter; this transformation is stopping and would close the iterator as soon
+ // as the count is observed; if that happens in the middle of an open RT, its end bound will not be included.
+ // If tracking repaired data, the counter is needed for overreading repaired data, otherwise we can
+ // optimise the case where this.limit = DataLimits.NONE which skips an unnecessary transform
+ if (executionController.isTrackingRepairedStatus())
+ {
+ DataLimits.Counter limit =
limits().newCounter(nowInSec(), false, selectsFullPartition(), metadata().enforceStrictLiveness());
- iterator = limit.applyTo(iterator);
- // ensure that a consistent amount of repaired data is read on each replica. This causes silent
- // overreading from the repaired data set, up to limits(). The extra data is not visible to
- // the caller, only iterated to produce the repaired data digest.
- iterator = executionController.getRepairedDataInfo().extend(iterator, limit);
- }
- else
- {
- iterator = limits().filter(iterator, nowInSec(), selectsFullPartition());
- }
+ iterator = limit.applyTo(iterator);
+ // ensure that a consistent amount of repaired data is read on each replica. This causes silent
+ // overreading from the repaired data set, up to limits(). The extra data is not visible to
+ // the caller, only iterated to produce the repaired data digest.
+ iterator = executionController.getRepairedDataInfo().extend(iterator, limit);
+ }
+ else
+ {
+ iterator = limits().filter(iterator, nowInSec(), selectsFullPartition());
+ }
- // because of the above, we need to append an aritifical end bound if the source iterator was stopped short by a counter.
- return RTBoundCloser.close(iterator);
+ // because of the above, we need to append an aritifical end bound if the source iterator was stopped short by a counter.
+ return RTBoundCloser.close(iterator);
+ }
+ catch (RuntimeException | Error e)
+ {
+ iterator.close();
+ throw e;
+ }
}
- catch (RuntimeException | Error e)
+ finally
{
- iterator.close();
- throw e;
+ COMMAND.set(null);
}
}
@@ -632,6 +652,88 @@ public abstract class ReadCommand extends AbstractReadQuery
}
}
+ private boolean shouldTrackSize(long warnThresholdBytes, long abortThresholdBytes)
+ {
+ return trackWarnings
+ && !SchemaConstants.isSystemKeyspace(metadata().keyspace)
+ && !(warnThresholdBytes == 0 && abortThresholdBytes == 0);
+ }
+
+ private UnfilteredPartitionIterator withQuerySizeTracking(UnfilteredPartitionIterator iterator)
+ {
+ final long warnThresholdBytes = DatabaseDescriptor.getLocalReadSizeWarnThresholdKb() * 1024;
+ final long abortThresholdBytes = DatabaseDescriptor.getLocalReadSizeAbortThresholdKb() * 1024;
+ if (!shouldTrackSize(warnThresholdBytes, abortThresholdBytes))
+ return iterator;
+ class QuerySizeTracking extends Transformation
+ {
+ private long sizeInBytes = 0;
+
+ @Override
+ public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator iter)
+ {
+ sizeInBytes += ObjectSizes.sizeOnHeapOf(iter.partitionKey().getKey());
+ return Transformation.apply(iter, this);
+ }
+
+ @Override
+ protected Row applyToStatic(Row row)
+ {
+ return applyToRow(row);
+ }
+
+ @Override
+ protected Row applyToRow(Row row)
+ {
+ addSize(row.unsharedHeapSize());
+ return row;
+ }
+
+ @Override
+ protected RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
+ {
+ addSize(marker.unsharedHeapSize());
+ return marker;
+ }
+
+ @Override
+ protected DeletionTime applyToDeletion(DeletionTime deletionTime)
+ {
+ addSize(deletionTime.unsharedHeapSize());
+ return deletionTime;
+ }
+
+ private void addSize(long size)
+ {
+ this.sizeInBytes += size;
+ if (abortThresholdBytes != 0 && this.sizeInBytes >= abortThresholdBytes)
+ {
+ String msg = String.format("Query %s attempted to read %d bytes but max allowed is %d; query aborted (see track_warnings.local_read_size.abort_threshold_kb)",
+ ReadCommand.this.toCQLString(), this.sizeInBytes, abortThresholdBytes);
+ Tracing.trace(msg);
+ MessageParams.remove(ParamType.LOCAL_READ_SIZE_WARN);
+ MessageParams.add(ParamType.LOCAL_READ_SIZE_ABORT, this.sizeInBytes);
+ throw new LocalReadSizeTooLargeException(msg);
+ }
+ else if (warnThresholdBytes != 0 && this.sizeInBytes >= warnThresholdBytes)
+ {
+ MessageParams.add(ParamType.LOCAL_READ_SIZE_WARN, this.sizeInBytes);
+ }
+ }
+
+ @Override
+ protected void onClose()
+ {
+ ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(metadata().id);
+ if (cfs != null)
+ cfs.metric.localReadSize.update(sizeInBytes);
+ }
+ }
+
+ iterator = Transformation.apply(iterator, new QuerySizeTracking());
+ return iterator;
+ }
+
protected UnfilteredPartitionIterator withStateTracking(UnfilteredPartitionIterator iter)
{
return Transformation.apply(iter, new CheckForAbort());
diff --git a/src/java/org/apache/cassandra/db/RowIndexEntry.java b/src/java/org/apache/cassandra/db/RowIndexEntry.java
index 215768bc8d..895bea9ab0 100644
--- a/src/java/org/apache/cassandra/db/RowIndexEntry.java
+++ b/src/java/org/apache/cassandra/db/RowIndexEntry.java
@@ -24,6 +24,7 @@ import java.util.List;
import com.codahale.metrics.Histogram;
import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.filter.RowIndexEntryTooLargeException;
import org.apache.cassandra.io.ISerializer;
import org.apache.cassandra.io.sstable.IndexInfo;
import org.apache.cassandra.io.sstable.format.Version;
@@ -36,6 +37,9 @@ import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.io.util.TrackedDataInputPlus;
import org.apache.cassandra.metrics.DefaultNameFactory;
import org.apache.cassandra.metrics.MetricNameFactory;
+import org.apache.cassandra.net.ParamType;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.vint.VIntCoding;
import org.github.jamm.Unmetered;
@@ -324,6 +328,8 @@ public class RowIndexEntry implements IMeasurableMemory
DeletionTime deletionTime = DeletionTime.serializer.deserialize(in);
int columnsIndexCount = (int) in.readUnsignedVInt();
+ checkSize(columnsIndexCount, size);
+
int indexedPartSize = size - serializedSize(deletionTime, headerLength, columnsIndexCount);
if (size <= DatabaseDescriptor.getColumnIndexCacheSize())
@@ -343,6 +349,51 @@ public class RowIndexEntry implements IMeasurableMemory
}
}
+ private void checkSize(int entries, int bytes)
+ {
+ ReadCommand command = ReadCommand.getCommand();
+ if (command == null || SchemaConstants.isSystemKeyspace(command.metadata().keyspace) || !DatabaseDescriptor.getTrackWarningsEnabled())
+ return;
+
+ int warnThreshold = DatabaseDescriptor.getRowIndexSizeWarnThresholdKb() * 1024;
+ int abortThreshold = DatabaseDescriptor.getRowIndexSizeAbortThresholdKb() * 1024;
+ if (warnThreshold == 0 && abortThreshold == 0)
+ return;
+
+ long estimatedMemory = estimateMaterializedIndexSize(entries, bytes);
+ ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(command.metadata().id);
+ if (cfs != null)
+ cfs.metric.rowIndexSize.update(estimatedMemory);
+
+ if (abortThreshold != 0 && estimatedMemory > abortThreshold)
+ {
+ String msg = String.format("Query %s attempted to access a large RowIndexEntry estimated to be %d bytes " +
+ "in-memory (total entries: %d, total bytes: %d) but the max allowed is %d;" +
+ " query aborted (see row_index_size_abort_threshold_kb)",
+ command.toCQLString(), estimatedMemory, entries, bytes, abortThreshold);
+ MessageParams.remove(ParamType.ROW_INDEX_SIZE_WARN);
+ MessageParams.add(ParamType.ROW_INDEX_SIZE_ABORT, estimatedMemory);
+
+ throw new RowIndexEntryTooLargeException(msg);
+ }
+ else if (warnThreshold != 0 && estimatedMemory > warnThreshold)
+ {
+ // use addIfLarger rather than add as a previous partition may be larger than this one
+ Long current = MessageParams.get(ParamType.ROW_INDEX_SIZE_WARN);
+ if (current == null || current.compareTo(estimatedMemory) < 0)
+ MessageParams.add(ParamType.ROW_INDEX_SIZE_WARN, estimatedMemory);
+ }
+ }
+
+ private static long estimateMaterializedIndexSize(int entries, int bytes)
+ {
+ long overhead = IndexInfo.EMPTY_SIZE
+ + ArrayClustering.EMPTY_SIZE
+ + DeletionTime.EMPTY_SIZE;
+
+ return (overhead * entries) + bytes;
+ }
+
public long deserializePositionAndSkip(DataInputPlus in) throws IOException
{
long position = in.readUnsignedVInt();
diff --git a/src/java/org/apache/cassandra/db/filter/LocalReadSizeTooLargeException.java b/src/java/org/apache/cassandra/db/filter/LocalReadSizeTooLargeException.java
new file mode 100644
index 0000000000..9d872dfa83
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/filter/LocalReadSizeTooLargeException.java
@@ -0,0 +1,29 @@
+/*
+ * 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.db.filter;
+
+import org.apache.cassandra.db.RejectException;
+
+public class LocalReadSizeTooLargeException extends RejectException
+{
+ public LocalReadSizeTooLargeException(String message)
+ {
+ super(message);
+ }
+}
diff --git a/src/java/org/apache/cassandra/db/filter/RowIndexEntryTooLargeException.java b/src/java/org/apache/cassandra/db/filter/RowIndexEntryTooLargeException.java
new file mode 100644
index 0000000000..5f7bfcdf0d
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/filter/RowIndexEntryTooLargeException.java
@@ -0,0 +1,29 @@
+/*
+ * 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.db.filter;
+
+import org.apache.cassandra.db.RejectException;
+
+public class RowIndexEntryTooLargeException extends RejectException
+{
+ public RowIndexEntryTooLargeException(String message)
+ {
+ super(message);
+ }
+}
diff --git a/src/java/org/apache/cassandra/db/rows/AbstractRangeTombstoneMarker.java b/src/java/org/apache/cassandra/db/rows/AbstractRangeTombstoneMarker.java
index 7dac1fae70..be328474ce 100644
--- a/src/java/org/apache/cassandra/db/rows/AbstractRangeTombstoneMarker.java
+++ b/src/java/org/apache/cassandra/db/rows/AbstractRangeTombstoneMarker.java
@@ -17,10 +17,8 @@
*/
package org.apache.cassandra.db.rows;
-import java.nio.ByteBuffer;
-
-import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.ClusteringBoundOrBoundary;
+import org.apache.cassandra.schema.TableMetadata;
public abstract class AbstractRangeTombstoneMarker> implements RangeTombstoneMarker
{
diff --git a/src/java/org/apache/cassandra/db/rows/ArrayCell.java b/src/java/org/apache/cassandra/db/rows/ArrayCell.java
index c4fdd14afb..eddc11ca0f 100644
--- a/src/java/org/apache/cassandra/db/rows/ArrayCell.java
+++ b/src/java/org/apache/cassandra/db/rows/ArrayCell.java
@@ -110,6 +110,13 @@ public class ArrayCell extends AbstractCell
return new BufferCell(column, timestamp, ttl, localDeletionTime, allocator.clone(value), path == null ? null : path.copy(allocator));
}
+ @Override
+ public long unsharedHeapSize()
+ {
+ return EMPTY_SIZE + ObjectSizes.sizeOfArray(value) + (path == null ? 0 : path.unsharedHeapSize());
+ }
+
+ @Override
public long unsharedHeapSizeExcludingData()
{
return EMPTY_SIZE + ObjectSizes.sizeOfEmptyByteArray() + (path == null ? 0 : path.unsharedHeapSizeExcludingData());
diff --git a/src/java/org/apache/cassandra/db/rows/BTreeRow.java b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
index bd44b666fd..2d3ee83de5 100644
--- a/src/java/org/apache/cassandra/db/rows/BTreeRow.java
+++ b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
@@ -497,6 +497,19 @@ public class BTreeRow extends AbstractRow
return Ints.checkedCast(accumulate((cd, v) -> v + cd.dataSize(), dataSize));
}
+ @Override
+ public long unsharedHeapSize()
+ {
+ long heapSize = EMPTY_SIZE
+ + clustering.unsharedHeapSize()
+ + primaryKeyLivenessInfo.unsharedHeapSize()
+ + deletion.unsharedHeapSize()
+ + BTree.sizeOfStructureOnHeap(btree);
+
+ return accumulate((cd, v) -> v + cd.unsharedHeapSize(), heapSize);
+ }
+
+ @Override
public long unsharedHeapSizeExcludingData()
{
long heapSize = EMPTY_SIZE
diff --git a/src/java/org/apache/cassandra/db/rows/BufferCell.java b/src/java/org/apache/cassandra/db/rows/BufferCell.java
index 55fc4b470d..7870bf100f 100644
--- a/src/java/org/apache/cassandra/db/rows/BufferCell.java
+++ b/src/java/org/apache/cassandra/db/rows/BufferCell.java
@@ -142,6 +142,13 @@ public class BufferCell extends AbstractCell
return new BufferCell(column, timestamp, ttl, localDeletionTime, allocator.clone(value), path == null ? null : path.copy(allocator));
}
+ @Override
+ public long unsharedHeapSize()
+ {
+ return EMPTY_SIZE + ObjectSizes.sizeOnHeapOf(value) + (path == null ? 0 : path.unsharedHeapSize());
+ }
+
+ @Override
public long unsharedHeapSizeExcludingData()
{
return EMPTY_SIZE + ObjectSizes.sizeOfEmptyHeapByteBuffer() + (path == null ? 0 : path.unsharedHeapSizeExcludingData());
diff --git a/src/java/org/apache/cassandra/db/rows/CellPath.java b/src/java/org/apache/cassandra/db/rows/CellPath.java
index 50496a1549..27b6272122 100644
--- a/src/java/org/apache/cassandra/db/rows/CellPath.java
+++ b/src/java/org/apache/cassandra/db/rows/CellPath.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
+import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.Digest;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.DataInputPlus;
@@ -31,7 +32,7 @@ import org.apache.cassandra.utils.memory.AbstractAllocator;
/**
* A path for a cell belonging to a complex column type (non-frozen collection or UDT).
*/
-public abstract class CellPath
+public abstract class CellPath implements IMeasurableMemory
{
public static final CellPath BOTTOM = new EmptyCellPath();
public static final CellPath TOP = new EmptyCellPath();
@@ -125,6 +126,13 @@ public abstract class CellPath
return new SingleItemCellPath(allocator.clone(value));
}
+ @Override
+ public long unsharedHeapSize()
+ {
+ return EMPTY_SIZE + ObjectSizes.sizeOnHeapOf(value);
+ }
+
+ @Override
public long unsharedHeapSizeExcludingData()
{
return EMPTY_SIZE + ObjectSizes.sizeOfEmptyHeapByteBuffer();
@@ -148,6 +156,14 @@ public abstract class CellPath
return this;
}
+ @Override
+ public long unsharedHeapSize()
+ {
+ // empty only happens with a cached reference, so 0 unshared space
+ return 0;
+ }
+
+ @Override
public long unsharedHeapSizeExcludingData()
{
return 0;
diff --git a/src/java/org/apache/cassandra/db/rows/ColumnData.java b/src/java/org/apache/cassandra/db/rows/ColumnData.java
index 36aad974cd..4146946616 100644
--- a/src/java/org/apache/cassandra/db/rows/ColumnData.java
+++ b/src/java/org/apache/cassandra/db/rows/ColumnData.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.db.rows;
import java.util.Comparator;
+import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.Digest;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.db.DeletionPurger;
@@ -31,7 +32,7 @@ import org.apache.cassandra.serializers.MarshalException;
* In practice, there is only 2 implementations of this: either {@link Cell} for simple columns
* or {@code ComplexColumnData} for complex columns.
*/
-public abstract class ColumnData
+public abstract class ColumnData implements IMeasurableMemory
{
public static final Comparator comparator = (cd1, cd2) -> cd1.column().compareTo(cd2.column());
diff --git a/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java b/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java
index 9f35437633..bf7714d9df 100644
--- a/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java
+++ b/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java
@@ -121,6 +121,16 @@ public class ComplexColumnData extends ColumnData implements Iterable>
return size;
}
+ @Override
+ public long unsharedHeapSize()
+ {
+ long heapSize = EMPTY_SIZE + ObjectSizes.sizeOfArray(cells);
+ for (Cell> cell : this)
+ heapSize += cell.unsharedHeapSize();
+ return heapSize;
+ }
+
+ @Override
public long unsharedHeapSizeExcludingData()
{
long heapSize = EMPTY_SIZE + ObjectSizes.sizeOfArray(cells);
diff --git a/src/java/org/apache/cassandra/db/rows/NativeCell.java b/src/java/org/apache/cassandra/db/rows/NativeCell.java
index 02e000823a..03dfc7090f 100644
--- a/src/java/org/apache/cassandra/db/rows/NativeCell.java
+++ b/src/java/org/apache/cassandra/db/rows/NativeCell.java
@@ -166,6 +166,13 @@ public class NativeCell extends AbstractCell
return new BufferCell(column, timestamp(), ttl(), localDeletionTime(), ByteBufferUtil.EMPTY_BYTE_BUFFER, path());
}
+ @Override
+ public long unsharedHeapSize()
+ {
+ return EMPTY_SIZE;
+ }
+
+ @Override
public long unsharedHeapSizeExcludingData()
{
return EMPTY_SIZE;
diff --git a/src/java/org/apache/cassandra/db/rows/RangeTombstoneBoundMarker.java b/src/java/org/apache/cassandra/db/rows/RangeTombstoneBoundMarker.java
index 20bc48415f..c38a6cd00a 100644
--- a/src/java/org/apache/cassandra/db/rows/RangeTombstoneBoundMarker.java
+++ b/src/java/org/apache/cassandra/db/rows/RangeTombstoneBoundMarker.java
@@ -22,6 +22,7 @@ import java.util.Objects;
import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.*;
+import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
/**
@@ -29,6 +30,8 @@ import org.apache.cassandra.utils.memory.AbstractAllocator;
*/
public class RangeTombstoneBoundMarker extends AbstractRangeTombstoneMarker>
{
+ private static final long EMPTY_SIZE = ObjectSizes.measure(new RangeTombstoneBoundMarker(new ArrayClusteringBound(ClusteringPrefix.Kind.INCL_START_BOUND, AbstractArrayClusteringPrefix.EMPTY_VALUES_ARRAY), null));
+
private final DeletionTime deletion;
public RangeTombstoneBoundMarker(ClusteringBound> bound, DeletionTime deletion)
@@ -156,6 +159,12 @@ public class RangeTombstoneBoundMarker extends AbstractRangeTombstoneMarker>
{
+ private static final long EMPTY_SIZE = ObjectSizes.measure(new RangeTombstoneBoundaryMarker(new ArrayClusteringBoundary(ClusteringPrefix.Kind.INCL_END_EXCL_START_BOUNDARY, new byte[][] { new byte[0]}), null, null));
+
private final DeletionTime endDeletion;
private final DeletionTime startDeletion;
@@ -187,6 +190,12 @@ public class RangeTombstoneBoundaryMarker extends AbstractRangeTombstoneMarker
* There is 2 types of markers: bounds (see {@link RangeTombstoneBoundMarker}) and boundaries (see {@link RangeTombstoneBoundaryMarker}).
*/
-public interface RangeTombstoneMarker extends Unfiltered
+public interface RangeTombstoneMarker extends Unfiltered, IMeasurableMemory
{
@Override
public ClusteringBoundOrBoundary> clustering();
diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java
index 5c28cd1b42..85d27a44eb 100644
--- a/src/java/org/apache/cassandra/db/rows/Row.java
+++ b/src/java/org/apache/cassandra/db/rows/Row.java
@@ -21,6 +21,7 @@ import java.util.*;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
+import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.schema.ColumnMetadata;
@@ -48,7 +49,7 @@ import org.apache.cassandra.utils.btree.UpdateFunction;
* it's own data. For instance, a {@code Row} cannot contains a cell that is deleted by its own
* row deletion.
*/
-public interface Row extends Unfiltered, Iterable
+public interface Row extends Unfiltered, Iterable, IMeasurableMemory
{
/**
* The clustering values for this row.
diff --git a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java
index 3f6c2d4653..f205900bf8 100644
--- a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java
+++ b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java
@@ -36,7 +36,7 @@ public enum RequestFailureReason
READ_TOO_MANY_TOMBSTONES (1),
TIMEOUT (2),
INCOMPATIBLE_SCHEMA (3),
- READ_TOO_LARGE (4);
+ READ_SIZE (4);
public static final Serializer serializer = new Serializer();
diff --git a/src/java/org/apache/cassandra/exceptions/TombstoneAbortException.java b/src/java/org/apache/cassandra/exceptions/TombstoneAbortException.java
index e86e760ae6..ef30ff3f49 100644
--- a/src/java/org/apache/cassandra/exceptions/TombstoneAbortException.java
+++ b/src/java/org/apache/cassandra/exceptions/TombstoneAbortException.java
@@ -23,14 +23,14 @@ import java.util.Map;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.locator.InetAddressAndPort;
-import static org.apache.cassandra.service.reads.ReadCallback.tombstoneAbortMessage;
+import static org.apache.cassandra.service.reads.trackwarnings.WarningsSnapshot.tombstoneAbortMessage;
public class TombstoneAbortException extends ReadAbortException
{
public final int nodes;
- public final int tombstones;
+ public final long tombstones;
- public TombstoneAbortException(int nodes, int tombstones, String cql, boolean dataPresent, ConsistencyLevel consistency, int received, int blockFor, Map failureReasonByEndpoint)
+ public TombstoneAbortException(int nodes, long tombstones, String cql, boolean dataPresent, ConsistencyLevel consistency, int received, int blockFor, Map failureReasonByEndpoint)
{
super(tombstoneAbortMessage(nodes, tombstones, cql), consistency, received, blockFor, dataPresent, failureReasonByEndpoint);
this.nodes = nodes;
diff --git a/src/java/org/apache/cassandra/io/sstable/IndexInfo.java b/src/java/org/apache/cassandra/io/sstable/IndexInfo.java
index e24436d017..e74415052a 100644
--- a/src/java/org/apache/cassandra/io/sstable/IndexInfo.java
+++ b/src/java/org/apache/cassandra/io/sstable/IndexInfo.java
@@ -27,7 +27,6 @@ import org.apache.cassandra.db.RowIndexEntry;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.ByteArrayAccessor;
import org.apache.cassandra.io.ISerializer;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.util.DataInputPlus;
@@ -59,7 +58,7 @@ import org.apache.cassandra.utils.ObjectSizes;
*/
public class IndexInfo
{
- private static final long EMPTY_SIZE = ObjectSizes.measure(new IndexInfo(null, null, 0, 0, null));
+ public static final long EMPTY_SIZE = ObjectSizes.measure(new IndexInfo(null, null, 0, 0, null));
public final long offset;
public final long width;
diff --git a/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java b/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java
index d0607af071..776027e395 100644
--- a/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java
@@ -156,8 +156,17 @@ public class KeyspaceMetrics
public final Meter clientTombstoneWarnings;
public final Meter clientTombstoneAborts;
- public final Meter clientReadSizeWarnings;
- public final Meter clientReadSizeAborts;
+ public final Meter coordinatorReadSizeWarnings;
+ public final Meter coordinatorReadSizeAborts;
+ public final Histogram coordinatorReadSize;
+
+ public final Meter localReadSizeWarnings;
+ public final Meter localReadSizeAborts;
+ public final Histogram localReadSize;
+
+ public final Meter rowIndexSizeWarnings;
+ public final Meter rowIndexSizeAborts;
+ public final Histogram rowIndexSize;
public final MetricNameFactory factory;
private Keyspace keyspace;
@@ -245,8 +254,17 @@ public class KeyspaceMetrics
clientTombstoneWarnings = createKeyspaceMeter("ClientTombstoneWarnings");
clientTombstoneAborts = createKeyspaceMeter("ClientTombstoneAborts");
- clientReadSizeWarnings = createKeyspaceMeter("ClientReadSizeWarnings");
- clientReadSizeAborts = createKeyspaceMeter("ClientReadSizeAborts");
+ coordinatorReadSizeWarnings = createKeyspaceMeter("CoordinatorReadSizeWarnings");
+ coordinatorReadSizeAborts = createKeyspaceMeter("CoordinatorReadSizeAborts");
+ coordinatorReadSize = createKeyspaceHistogram("CoordinatorReadSize", false);
+
+ localReadSizeWarnings = createKeyspaceMeter("LocalReadSizeWarnings");
+ localReadSizeAborts = createKeyspaceMeter("LocalReadSizeAborts");
+ localReadSize = createKeyspaceHistogram("LocalReadSize", false);
+
+ rowIndexSizeWarnings = createKeyspaceMeter("RowIndexSizeWarnings");
+ rowIndexSizeAborts = createKeyspaceMeter("RowIndexSizeAborts");
+ rowIndexSize = createKeyspaceHistogram("RowIndexSize", false);
}
/**
diff --git a/src/java/org/apache/cassandra/metrics/TableMetrics.java b/src/java/org/apache/cassandra/metrics/TableMetrics.java
index ced062225b..6b7193c6c7 100644
--- a/src/java/org/apache/cassandra/metrics/TableMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/TableMetrics.java
@@ -261,8 +261,18 @@ public class TableMetrics
public final TableMeter clientTombstoneWarnings;
public final TableMeter clientTombstoneAborts;
- public final TableMeter clientReadSizeWarnings;
- public final TableMeter clientReadSizeAborts;
+
+ public final TableMeter coordinatorReadSizeWarnings;
+ public final TableMeter coordinatorReadSizeAborts;
+ public final TableHistogram coordinatorReadSize;
+
+ public final TableMeter localReadSizeWarnings;
+ public final TableMeter localReadSizeAborts;
+ public final TableHistogram localReadSize;
+
+ public final TableMeter rowIndexSizeWarnings;
+ public final TableMeter rowIndexSizeAborts;
+ public final TableHistogram rowIndexSize;
private static Pair totalNonSystemTablesSize(Predicate predicate)
{
@@ -922,8 +932,17 @@ public class TableMetrics
clientTombstoneWarnings = createTableMeter("ClientTombstoneWarnings", cfs.keyspace.metric.clientTombstoneWarnings);
clientTombstoneAborts = createTableMeter("ClientTombstoneAborts", cfs.keyspace.metric.clientTombstoneAborts);
- clientReadSizeWarnings = createTableMeter("ClientReadSizeWarnings", cfs.keyspace.metric.clientReadSizeWarnings);
- clientReadSizeAborts = createTableMeter("ClientReadSizeAborts", cfs.keyspace.metric.clientReadSizeAborts);
+ coordinatorReadSizeWarnings = createTableMeter("CoordinatorReadSizeWarnings", cfs.keyspace.metric.coordinatorReadSizeWarnings);
+ coordinatorReadSizeAborts = createTableMeter("CoordinatorReadSizeAborts", cfs.keyspace.metric.coordinatorReadSizeAborts);
+ coordinatorReadSize = createTableHistogram("CoordinatorReadSize", cfs.keyspace.metric.coordinatorReadSize, false);
+
+ localReadSizeWarnings = createTableMeter("LocalReadSizeWarnings", cfs.keyspace.metric.localReadSizeWarnings);
+ localReadSizeAborts = createTableMeter("LocalReadSizeAborts", cfs.keyspace.metric.localReadSizeAborts);
+ localReadSize = createTableHistogram("LocalReadSize", cfs.keyspace.metric.localReadSize, false);
+
+ rowIndexSizeWarnings = createTableMeter("RowIndexSizeWarnings", cfs.keyspace.metric.rowIndexSizeWarnings);
+ rowIndexSizeAborts = createTableMeter("RowIndexSizeAborts", cfs.keyspace.metric.rowIndexSizeAborts);
+ rowIndexSize = createTableHistogram("RowIndexSize", cfs.keyspace.metric.rowIndexSize, false);
}
public void updateSSTableIterated(int count)
diff --git a/src/java/org/apache/cassandra/net/ParamType.java b/src/java/org/apache/cassandra/net/ParamType.java
index d8b8c0ed2c..038530bbc6 100644
--- a/src/java/org/apache/cassandra/net/ParamType.java
+++ b/src/java/org/apache/cassandra/net/ParamType.java
@@ -25,6 +25,7 @@ import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.Int32Serializer;
+import org.apache.cassandra.utils.Int64Serializer;
import org.apache.cassandra.utils.UUIDSerializer;
import static java.lang.Math.max;
@@ -57,7 +58,11 @@ public enum ParamType
TRACK_REPAIRED_DATA (7, "TrackRepaired", LegacyFlag.serializer),
TOMBSTONE_ABORT(8, "TSA", Int32Serializer.serializer),
- TOMBSTONE_WARNING(9, "TSW", Int32Serializer.serializer);
+ TOMBSTONE_WARNING(9, "TSW", Int32Serializer.serializer),
+ LOCAL_READ_SIZE_ABORT(10, "LRSA", Int64Serializer.serializer),
+ LOCAL_READ_SIZE_WARN(11, "LRSW", Int64Serializer.serializer),
+ ROW_INDEX_SIZE_ABORT(12, "RISA", Int64Serializer.serializer),
+ ROW_INDEX_SIZE_WARN(13, "RISW", Int64Serializer.serializer);
final int id;
@Deprecated final String legacyAlias; // pre-4.0 we used to serialize entire param name string
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index dcd2598a35..0e663c02a9 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -6128,41 +6128,105 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
@Override
- public long getClientLargeReadWarnThresholdKB()
+ public boolean getTrackWarningsEnabled()
{
- return DatabaseDescriptor.getClientLargeReadWarnThresholdKB();
+ return DatabaseDescriptor.getTrackWarningsEnabled();
}
@Override
- public void setClientLargeReadWarnThresholdKB(long threshold)
+ public void setTrackWarningsEnabled(boolean value)
{
- DatabaseDescriptor.setClientLargeReadWarnThresholdKB(threshold);
- logger.info("updated client_large_read_warn_threshold_kb to {}", threshold);
+ DatabaseDescriptor.setTrackWarningsEnabled(value);
+ logger.info("updated track_warnings.enabled to {}", value);
}
@Override
- public long getClientLargeReadAbortThresholdKB()
+ public long getCoordinatorLargeReadWarnThresholdKB()
{
- return DatabaseDescriptor.getClientLargeReadAbortThresholdKB();
+ return DatabaseDescriptor.getCoordinatorReadSizeWarnThresholdKB();
}
@Override
- public void setClientLargeReadAbortThresholdKB(long threshold)
+ public void setCoordinatorLargeReadWarnThresholdKB(long threshold)
{
- DatabaseDescriptor.setClientLargeReadAbortThresholdKB(threshold);
- logger.info("updated client_large_read_abort_threshold_kb to {}", threshold);
+ if (threshold < 0)
+ throw new IllegalArgumentException("threshold " + threshold + " is less than 0; must be positive or zero");
+ DatabaseDescriptor.setCoordinatorReadSizeWarnThresholdKB(threshold);
+ logger.info("updated track_warnings.coordinator_large_read.warn_threshold_kb to {}", threshold);
}
@Override
- public boolean getClientTrackWarningsEnabled()
+ public long getCoordinatorLargeReadAbortThresholdKB()
{
- return DatabaseDescriptor.getClientTrackWarningsEnabled();
+ return DatabaseDescriptor.getCoordinatorReadSizeAbortThresholdKB();
}
@Override
- public void setClientTrackWarningsEnabled(boolean value)
+ public void setCoordinatorLargeReadAbortThresholdKB(long threshold)
{
- DatabaseDescriptor.setClientTrackWarningsEnabled(value);
- logger.info("updated client_track_warnings_enabled to {}", value);
+ if (threshold < 0)
+ throw new IllegalArgumentException("threshold " + threshold + " is less than 0; must be positive or zero");
+ DatabaseDescriptor.setCoordinatorReadSizeAbortThresholdKB(threshold);
+ logger.info("updated track_warnings.coordinator_large_read.abort_threshold_kb to {}", threshold);
+ }
+
+ @Override
+ public long getLocalReadTooLargeWarnThresholdKb()
+ {
+ return DatabaseDescriptor.getLocalReadSizeWarnThresholdKb();
+ }
+
+ @Override
+ public void setLocalReadTooLargeWarnThresholdKb(long value)
+ {
+ if (value < 0)
+ throw new IllegalArgumentException("value " + value + " is less than 0; must be positive or zero");
+ DatabaseDescriptor.setLocalReadSizeWarnThresholdKb(value);
+ logger.info("updated track_warnings.local_read_size.warn_threshold_kb to {}", value);
+ }
+
+ @Override
+ public long getLocalReadTooLargeAbortThresholdKb()
+ {
+ return DatabaseDescriptor.getLocalReadSizeAbortThresholdKb();
+ }
+
+ @Override
+ public void setLocalReadTooLargeAbortThresholdKb(long value)
+ {
+ if (value < 0)
+ throw new IllegalArgumentException("value " + value + " is less than 0; must be positive or zero");
+ DatabaseDescriptor.setLocalReadSizeAbortThresholdKb(value);
+ logger.info("updated track_warnings.local_read_size.abort_threshold_kb to {}", value);
+ }
+
+ @Override
+ public int getRowIndexSizeWarnThresholdKb()
+ {
+ return DatabaseDescriptor.getRowIndexSizeWarnThresholdKb();
+ }
+
+ @Override
+ public void setRowIndexSizeWarnThresholdKb(int value)
+ {
+ if (value < 0)
+ throw new IllegalArgumentException("value " + value + " is less than 0; must be positive or zero");
+ DatabaseDescriptor.setRowIndexSizeWarnThresholdKb(value);
+ logger.info("updated track_warnings.row_index_size.warn_threshold_kb to {}", value);
+ }
+
+ @Override
+ public int getRowIndexSizeAbortThresholdKb()
+ {
+ return DatabaseDescriptor.getRowIndexSizeAbortThresholdKb();
+ }
+
+ @Override
+ public void setRowIndexSizeAbortThresholdKb(int value)
+ {
+ if (value < 0)
+ throw new IllegalArgumentException("value " + value + " is less than 0; must be positive or zero");
+ DatabaseDescriptor.setRowIndexSizeAbortThresholdKb(value);
+ logger.info("updated track_warnings.row_index_size.abort_threshold_kb to {}", value);
}
}
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index f29dd892eb..6a294c1bb2 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -892,10 +892,21 @@ public interface StorageServiceMBean extends NotificationEmitter
public void setCompactionTombstoneWarningThreshold(int count);
public int getCompactionTombstoneWarningThreshold();
- public long getClientLargeReadWarnThresholdKB();
- public void setClientLargeReadWarnThresholdKB(long threshold);
- public long getClientLargeReadAbortThresholdKB();
- public void setClientLargeReadAbortThresholdKB(long threshold);
- public boolean getClientTrackWarningsEnabled();
- public void setClientTrackWarningsEnabled(boolean value);
+ public boolean getTrackWarningsEnabled();
+ public void setTrackWarningsEnabled(boolean value);
+
+ public long getCoordinatorLargeReadWarnThresholdKB();
+ public void setCoordinatorLargeReadWarnThresholdKB(long threshold);
+ public long getCoordinatorLargeReadAbortThresholdKB();
+ public void setCoordinatorLargeReadAbortThresholdKB(long threshold);
+
+ public long getLocalReadTooLargeWarnThresholdKb();
+ public void setLocalReadTooLargeWarnThresholdKb(long value);
+ public long getLocalReadTooLargeAbortThresholdKb();
+ public void setLocalReadTooLargeAbortThresholdKb(long value);
+
+ public int getRowIndexSizeWarnThresholdKb();
+ public void setRowIndexSizeWarnThresholdKb(int value);
+ public int getRowIndexSizeAbortThresholdKb();
+ public void setRowIndexSizeAbortThresholdKb(int value);
}
diff --git a/src/java/org/apache/cassandra/service/reads/ReadCallback.java b/src/java/org/apache/cassandra/service/reads/ReadCallback.java
index 67031478a5..15f1559d04 100644
--- a/src/java/org/apache/cassandra/service/reads/ReadCallback.java
+++ b/src/java/org/apache/cassandra/service/reads/ReadCallback.java
@@ -20,20 +20,14 @@ package org.apache.cassandra.service.reads;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
-import com.google.common.annotations.VisibleForTesting;
-
-import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.ColumnFamilyStore;
-import org.apache.cassandra.db.MessageParams;
-import org.apache.cassandra.exceptions.TombstoneAbortException;
-import org.apache.cassandra.locator.ReplicaPlan;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.MessageParams;
import org.apache.cassandra.db.PartitionRangeReadCommand;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadResponse;
@@ -42,12 +36,14 @@ import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.locator.Endpoints;
import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.ReplicaPlan;
+import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.ParamType;
import org.apache.cassandra.net.RequestCallback;
-import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.Verb;
-import org.apache.cassandra.schema.Schema;
-import org.apache.cassandra.service.ClientWarn;
+import org.apache.cassandra.service.reads.trackwarnings.CoordinatorWarnings;
+import org.apache.cassandra.service.reads.trackwarnings.WarningContext;
+import org.apache.cassandra.service.reads.trackwarnings.WarningsSnapshot;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
@@ -56,31 +52,6 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
public class ReadCallback, P extends ReplicaPlan.ForRead> implements RequestCallback
{
protected static final Logger logger = LoggerFactory.getLogger( ReadCallback.class );
- private class WarningCounter
- {
- // the highest number of tombstones reported by a node's warning
- final AtomicInteger tombstoneWarnings = new AtomicInteger();
- final AtomicInteger maxTombstoneWarningCount = new AtomicInteger();
- // the highest number of tombstones reported by a node's rejection. This should be the same as
- // our configured limit, but including to aid in diagnosing misconfigurations
- final AtomicInteger tombstoneAborts = new AtomicInteger();
- final AtomicInteger maxTombstoneAbortsCount = new AtomicInteger();
-
- // TODO: take message as arg and return boolean for 'had warning' etc
- void addTombstoneWarning(InetAddressAndPort from, int tombstones)
- {
- if (!waitingFor(from)) return;
- tombstoneWarnings.incrementAndGet();
- maxTombstoneWarningCount.accumulateAndGet(tombstones, Math::max);
- }
-
- void addTombstoneAbort(InetAddressAndPort from, int tombstones)
- {
- if (!waitingFor(from)) return;
- tombstoneAborts.incrementAndGet();
- maxTombstoneAbortsCount.accumulateAndGet(tombstones, Math::max);
- }
- }
public final ResponseResolver resolver;
final SimpleCondition condition = new SimpleCondition();
@@ -94,9 +65,9 @@ public class ReadCallback, P extends ReplicaPlan.ForRead<
= AtomicIntegerFieldUpdater.newUpdater(ReadCallback.class, "failures");
private volatile int failures = 0;
private final Map failureReasonByEndpoint;
- private volatile WarningCounter warningCounter;
- private static final AtomicReferenceFieldUpdater warningsUpdater
- = AtomicReferenceFieldUpdater.newUpdater(ReadCallback.class, ReadCallback.WarningCounter.class, "warningCounter");
+ private volatile WarningContext warningContext;
+ private static final AtomicReferenceFieldUpdater warningsUpdater
+ = AtomicReferenceFieldUpdater.newUpdater(ReadCallback.class, WarningContext.class, "warningContext");
public ReadCallback(ResponseResolver resolver, ReadCommand command, ReplicaPlan.Shared replicaPlan, long queryStartNanoTime)
{
@@ -131,23 +102,6 @@ public class ReadCallback, P extends ReplicaPlan.ForRead<
}
}
- @VisibleForTesting
- public static String tombstoneAbortMessage(int nodes, int tombstones, String cql)
- {
- return String.format("%s nodes scanned over %s tombstones and aborted the query %s (see tombstone_failure_threshold)", nodes, tombstones, cql);
- }
-
- @VisibleForTesting
- public static String tombstoneWarnMessage(int nodes, int tombstones, String cql)
- {
- return String.format("%s nodes scanned up to %s tombstones and issued tombstone warnings for query %s (see tombstone_warn_threshold)", nodes, tombstones, cql);
- }
-
- private ColumnFamilyStore cfs()
- {
- return Schema.instance.getColumnFamilyStoreInstance(command.metadata().id);
- }
-
public void awaitResults() throws ReadFailureException, ReadTimeoutException
{
boolean signaled = await(command.getTimeout(MILLISECONDS), TimeUnit.MILLISECONDS);
@@ -160,24 +114,17 @@ public class ReadCallback, P extends ReplicaPlan.ForRead<
*/
int received = resolver.responses.size();
boolean failed = failures > 0 && (blockFor > received || !resolver.isDataPresent());
- WarningCounter warnings = warningCounter;
+ WarningContext warnings = warningContext;
+ // save the snapshot so abort state is not changed between now and when mayAbort gets called
+ WarningsSnapshot snapshot = null;
if (warnings != null)
{
- if (warnings.tombstoneAborts.get() > 0)
- {
- String msg = tombstoneAbortMessage(warnings.tombstoneAborts.get(), warnings.maxTombstoneAbortsCount.get(), command.toCQLString());
- ClientWarn.instance.warn(msg + " with " + command.loggableTokens());
- logger.warn(msg);
- cfs().metric.clientTombstoneAborts.mark();
- }
-
- if (warnings.tombstoneWarnings.get() > 0)
- {
- String msg = tombstoneWarnMessage(warnings.tombstoneWarnings.get(), warnings.maxTombstoneWarningCount.get(), command.toCQLString());
- ClientWarn.instance.warn(msg + " with " + command.loggableTokens());
- logger.warn(msg);
- cfs().metric.clientTombstoneWarnings.mark();
- }
+ snapshot = warnings.snapshot();
+ // this is possible due to a race condition between waiting and responding
+ // network thread creates the WarningContext to update metrics, but we are actively reading and see it is empty
+ // this is likely to happen when a timeout happens or from a speculative response
+ if (!snapshot.isEmpty())
+ CoordinatorWarnings.update(command, snapshot);
}
if (signaled && !failed)
return;
@@ -193,9 +140,8 @@ public class ReadCallback, P extends ReplicaPlan.ForRead<
logger.debug("{}; received {} of {} responses{}", failed ? "Failed" : "Timed out", received, blockFor, gotData);
}
- if (warnings != null && warnings.tombstoneAborts.get() > 0)
- throw new TombstoneAbortException(warnings.tombstoneAborts.get(), warnings.maxTombstoneAbortsCount.get(), command.toCQLString(), resolver.isDataPresent(),
- replicaPlan.get().consistencyLevel(), received, blockFor, failureReasonByEndpoint);
+ if (snapshot != null)
+ snapshot.maybeAbort(command, replicaPlan().consistencyLevel(), received, blockFor, resolver.isDataPresent(), failureReasonByEndpoint);
// Same as for writes, see AbstractWriteResponseHandler
throw failed
@@ -213,15 +159,15 @@ public class ReadCallback, P extends ReplicaPlan.ForRead<
{
assertWaitingFor(message.from());
Map params = message.header.params();
- if (params.containsKey(ParamType.TOMBSTONE_ABORT))
+ InetAddressAndPort from = message.from();
+ if (WarningContext.isSupported(params.keySet()))
{
- getWarningCounter().addTombstoneAbort(message.from(), (Integer) params.get(ParamType.TOMBSTONE_ABORT));
- onFailure(message.from(), RequestFailureReason.READ_TOO_MANY_TOMBSTONES);
- return;
- }
- else if (params.containsKey(ParamType.TOMBSTONE_WARNING))
- {
- getWarningCounter().addTombstoneWarning(message.from(), (Integer) params.get(ParamType.TOMBSTONE_WARNING));
+ RequestFailureReason reason = getWarningContext().updateCounters(params, from);
+ if (reason != null)
+ {
+ onFailure(message.from(), reason);
+ return;
+ }
}
resolver.preprocess(message);
@@ -235,16 +181,16 @@ public class ReadCallback, P extends ReplicaPlan.ForRead<
condition.signalAll();
}
- private WarningCounter getWarningCounter()
+ private WarningContext getWarningContext()
{
- WarningCounter current;
+ WarningContext current;
do {
- current = warningCounter;
+ current = warningContext;
if (current != null)
return current;
- current = new WarningCounter();
+ current = new WarningContext();
} while (!warningsUpdater.compareAndSet(this, null, current));
return current;
}
@@ -286,12 +232,8 @@ public class ReadCallback, P extends ReplicaPlan.ForRead<
*/
private void assertWaitingFor(InetAddressAndPort from)
{
- assert waitingFor(from): "Received read response from unexpected replica: " + from;
- }
-
- private boolean waitingFor(InetAddressAndPort from)
- {
- return !replicaPlan().consistencyLevel().isDatacenterLocal()
- || DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(from));
+ assert !replicaPlan().consistencyLevel().isDatacenterLocal()
+ || DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(from))
+ : "Received read response from unexpected replica: " + from;
}
}
diff --git a/src/java/org/apache/cassandra/service/reads/trackwarnings/CoordinatorWarnings.java b/src/java/org/apache/cassandra/service/reads/trackwarnings/CoordinatorWarnings.java
new file mode 100644
index 0000000000..076c8165ca
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/reads/trackwarnings/CoordinatorWarnings.java
@@ -0,0 +1,198 @@
+/*
+ * 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.service.reads.trackwarnings;
+
+import java.util.AbstractMap;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.util.concurrent.FastThreadLocal;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.metrics.TableMetrics;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.service.ClientWarn;
+import org.apache.cassandra.service.reads.ReadCallback;
+
+public class CoordinatorWarnings
+{
+ private static final Logger logger = LoggerFactory.getLogger(CoordinatorWarnings.class);
+ private static final boolean ENABLE_DEFENSIVE_CHECKS = Boolean.getBoolean("cassandra.track_warnings.coordinator.defensive_checks_enabled");
+
+ // when .init() is called set the STATE to be INIT; this is to lazy allocate the map only when warnings are generated
+ private static final Map INIT = Collections.emptyMap();
+ private static final FastThreadLocal