diff --git a/CHANGES.txt b/CHANGES.txt index 4a9ae14c58..e74f33006a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -7,6 +7,7 @@ Merged from 3.11: * stop_paranoid disk failure policy is ignored on CorruptSSTableException after node is up (CASSANDRA-15191) * Frozen RawTuple is not annotated with frozen in the toString method (CASSANDRA-15857) Merged from 3.0: + * Operational improvements and hardening for replica filtering protection (CASSANDRA-15907) * Fix empty/null json string representation (CASSANDRA-15896) Merged from 2.2: * Fix CQL parsing of collections when the column type is reversed (CASSANDRA-15814) diff --git a/build.xml b/build.xml index f0948aef08..d861920d69 100644 --- a/build.xml +++ b/build.xml @@ -563,8 +563,7 @@ - - + diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 573c9367d4..fcd2ffa1b7 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -1226,6 +1226,26 @@ transparent_data_encryption_options: tombstone_warn_threshold: 1000 tombstone_failure_threshold: 100000 +# Filtering and secondary index queries at read consistency levels above ONE/LOCAL_ONE use a +# mechanism called replica filtering protection to ensure that results from stale replicas do +# not violate consistency. (See CASSANDRA-8272 and CASSANDRA-15907 for more details.) This +# mechanism materializes replica results by partition on-heap at the coordinator. The more possibly +# stale results returned by the replicas, the more rows materialized during the query. +replica_filtering_protection: + # These thresholds exist to limit the damage severely out-of-date replicas can cause during these + # queries. They limit the number of rows from all replicas individual index and filtering queries + # can materialize on-heap to return correct results at the desired read consistency level. + # + # "cached_replica_rows_warn_threshold" is the per-query threshold at which a warning will be logged. + # "cached_replica_rows_fail_threshold" is the per-query threshold at which the query will fail. + # + # These thresholds may also be adjusted at runtime using the StorageService mbean. + # + # If the failure threshold is breached, it is likely that either the current page/fetch size + # is too large or one or more replicas is severely out-of-sync and in need of repair. + cached_rows_warn_threshold: 2000 + cached_rows_fail_threshold: 32000 + # Log WARN on any multiple-partition batch size exceeding this value. 5kb per batch by default. # Caution should be taken on increasing the size of this threshold as it can lead to node instability. batch_size_warn_threshold_in_kb: 5 diff --git a/doc/source/operating/metrics.rst b/doc/source/operating/metrics.rst index fc37440d3c..0053cbcc7e 100644 --- a/doc/source/operating/metrics.rst +++ b/doc/source/operating/metrics.rst @@ -79,76 +79,80 @@ Reported name format: **all** tables and keyspaces on the node. -======================================= ============== =========== -Name Type Description -======================================= ============== =========== -MemtableOnHeapSize Gauge Total amount of data stored in the memtable that resides **on**-heap, including column related overhead and partitions overwritten. -MemtableOffHeapSize Gauge Total amount of data stored in the memtable that resides **off**-heap, including column related overhead and partitions overwritten. -MemtableLiveDataSize Gauge Total amount of live data stored in the memtable, excluding any data structure overhead. -AllMemtablesOnHeapSize Gauge Total amount of data stored in the memtables (2i and pending flush memtables included) that resides **on**-heap. -AllMemtablesOffHeapSize Gauge Total amount of data stored in the memtables (2i and pending flush memtables included) that resides **off**-heap. -AllMemtablesLiveDataSize Gauge Total amount of live data stored in the memtables (2i and pending flush memtables included) that resides off-heap, excluding any data structure overhead. -MemtableColumnsCount Gauge Total number of columns present in the memtable. -MemtableSwitchCount Counter Number of times flush has resulted in the memtable being switched out. -CompressionRatio Gauge Current compression ratio for all SSTables. -EstimatedPartitionSizeHistogram Gauge Histogram of estimated partition size (in bytes). -EstimatedPartitionCount Gauge Approximate number of keys in table. -EstimatedColumnCountHistogram Gauge Histogram of estimated number of columns. -SSTablesPerReadHistogram Histogram Histogram of the number of sstable data files accessed per single partition read. SSTables skipped due to Bloom Filters, min-max key or partition index lookup are not taken into acoount. -ReadLatency Latency Local read latency for this table. -RangeLatency Latency Local range scan latency for this table. -WriteLatency Latency Local write latency for this table. -CoordinatorReadLatency Timer Coordinator read latency for this table. -CoordinatorWriteLatency Timer Coordinator write latency for this table. -CoordinatorScanLatency Timer Coordinator range scan latency for this table. -PendingFlushes Counter Estimated number of flush tasks pending for this table. -BytesFlushed Counter Total number of bytes flushed since server [re]start. -CompactionBytesWritten Counter Total number of bytes written by compaction since server [re]start. -PendingCompactions Gauge Estimate of number of pending compactions for this table. -LiveSSTableCount Gauge Number of SSTables on disk for this table. -LiveDiskSpaceUsed Counter Disk space used by SSTables belonging to this table (in bytes). -TotalDiskSpaceUsed Counter Total disk space used by SSTables belonging to this table, including obsolete ones waiting to be GC'd. -MinPartitionSize Gauge Size of the smallest compacted partition (in bytes). -MaxPartitionSize Gauge Size of the largest compacted partition (in bytes). -MeanPartitionSize Gauge Size of the average compacted partition (in bytes). -BloomFilterFalsePositives Gauge Number of false positives on table's bloom filter. -BloomFilterFalseRatio Gauge False positive ratio of table's bloom filter. -BloomFilterDiskSpaceUsed Gauge Disk space used by bloom filter (in bytes). -BloomFilterOffHeapMemoryUsed Gauge Off-heap memory used by bloom filter. -IndexSummaryOffHeapMemoryUsed Gauge Off-heap memory used by index summary. -CompressionMetadataOffHeapMemoryUsed Gauge Off-heap memory used by compression meta data. -KeyCacheHitRate Gauge Key cache hit rate for this table. -TombstoneScannedHistogram Histogram Histogram of tombstones scanned in queries on this table. -LiveScannedHistogram Histogram Histogram of live cells scanned in queries on this table. -ColUpdateTimeDeltaHistogram Histogram Histogram of column update time delta on this table. -ViewLockAcquireTime Timer Time taken acquiring a partition lock for materialized view updates on this table. -ViewReadTime Timer Time taken during the local read of a materialized view update. -TrueSnapshotsSize Gauge Disk space used by snapshots of this table including all SSTable components. -RowCacheHitOutOfRange Counter Number of table row cache hits that do not satisfy the query filter, thus went to disk. -RowCacheHit Counter Number of table row cache hits. -RowCacheMiss Counter Number of table row cache misses. -CasPrepare Latency Latency of paxos prepare round. -CasPropose Latency Latency of paxos propose round. -CasCommit Latency Latency of paxos commit round. -PercentRepaired Gauge Percent of table data that is repaired on disk. -BytesRepaired Gauge Size of table data repaired on disk -BytesUnrepaired Gauge Size of table data unrepaired on disk -BytesPendingRepair Gauge Size of table data isolated for an ongoing incremental repair -SpeculativeRetries Counter Number of times speculative retries were sent for this table. -SpeculativeFailedRetries Counter Number of speculative retries that failed to prevent a timeout -SpeculativeInsufficientReplicas Counter Number of speculative retries that couldn't be attempted due to lack of replicas -SpeculativeSampleLatencyNanos Gauge Number of nanoseconds to wait before speculation is attempted. Value may be statically configured or updated periodically based on coordinator latency. -WaitingOnFreeMemtableSpace Histogram Histogram of time spent waiting for free memtable space, either on- or off-heap. -DroppedMutations Counter Number of dropped mutations on this table. -AnticompactionTime Timer Time spent anticompacting before a consistent repair. -ValidationTime Timer Time spent doing validation compaction during repair. -SyncTime Timer Time spent doing streaming during repair. -BytesValidated Histogram Histogram over the amount of bytes read during validation. -PartitionsValidated Histogram Histogram over the number of partitions read during validation. -BytesAnticompacted Counter How many bytes we anticompacted. -BytesMutatedAnticompaction Counter How many bytes we avoided anticompacting because the sstable was fully contained in the repaired range. -MutatedAnticompactionGauge Gauge Ratio of bytes mutated vs total bytes repaired. -======================================= ============== =========== +=============================================== ============== =========== +Name Type Description +=============================================== ============== =========== +MemtableOnHeapSize Gauge Total amount of data stored in the memtable that resides **on**-heap, including column related overhead and partitions overwritten. +MemtableOffHeapSize Gauge Total amount of data stored in the memtable that resides **off**-heap, including column related overhead and partitions overwritten. +MemtableLiveDataSize Gauge Total amount of live data stored in the memtable, excluding any data structure overhead. +AllMemtablesOnHeapSize Gauge Total amount of data stored in the memtables (2i and pending flush memtables included) that resides **on**-heap. +AllMemtablesOffHeapSize Gauge Total amount of data stored in the memtables (2i and pending flush memtables included) that resides **off**-heap. +AllMemtablesLiveDataSize Gauge Total amount of live data stored in the memtables (2i and pending flush memtables included) that resides off-heap, excluding any data structure overhead. +MemtableColumnsCount Gauge Total number of columns present in the memtable. +MemtableSwitchCount Counter Number of times flush has resulted in the memtable being switched out. +CompressionRatio Gauge Current compression ratio for all SSTables. +EstimatedPartitionSizeHistogram Gauge Histogram of estimated partition size (in bytes). +EstimatedPartitionCount Gauge Approximate number of keys in table. +EstimatedColumnCountHistogram Gauge Histogram of estimated number of columns. +SSTablesPerReadHistogram Histogram Histogram of the number of sstable data files accessed per single partition read. SSTables skipped due to Bloom Filters, min-max key or partition index lookup are not taken into acoount. +ReadLatency Latency Local read latency for this table. +RangeLatency Latency Local range scan latency for this table. +WriteLatency Latency Local write latency for this table. +CoordinatorReadLatency Timer Coordinator read latency for this table. +CoordinatorWriteLatency Timer Coordinator write latency for this table. +CoordinatorScanLatency Timer Coordinator range scan latency for this table. +PendingFlushes Counter Estimated number of flush tasks pending for this table. +BytesFlushed Counter Total number of bytes flushed since server [re]start. +CompactionBytesWritten Counter Total number of bytes written by compaction since server [re]start. +PendingCompactions Gauge Estimate of number of pending compactions for this table. +LiveSSTableCount Gauge Number of SSTables on disk for this table. +LiveDiskSpaceUsed Counter Disk space used by SSTables belonging to this table (in bytes). +TotalDiskSpaceUsed Counter Total disk space used by SSTables belonging to this table, including obsolete ones waiting to be GC'd. +MinPartitionSize Gauge Size of the smallest compacted partition (in bytes). +MaxPartitionSize Gauge Size of the largest compacted partition (in bytes). +MeanPartitionSize Gauge Size of the average compacted partition (in bytes). +BloomFilterFalsePositives Gauge Number of false positives on table's bloom filter. +BloomFilterFalseRatio Gauge False positive ratio of table's bloom filter. +BloomFilterDiskSpaceUsed Gauge Disk space used by bloom filter (in bytes). +BloomFilterOffHeapMemoryUsed Gauge Off-heap memory used by bloom filter. +IndexSummaryOffHeapMemoryUsed Gauge Off-heap memory used by index summary. +CompressionMetadataOffHeapMemoryUsed Gauge Off-heap memory used by compression meta data. +KeyCacheHitRate Gauge Key cache hit rate for this table. +TombstoneScannedHistogram Histogram Histogram of tombstones scanned in queries on this table. +LiveScannedHistogram Histogram Histogram of live cells scanned in queries on this table. +ColUpdateTimeDeltaHistogram Histogram Histogram of column update time delta on this table. +ViewLockAcquireTime Timer Time taken acquiring a partition lock for materialized view updates on this table. +ViewReadTime Timer Time taken during the local read of a materialized view update. +TrueSnapshotsSize Gauge Disk space used by snapshots of this table including all SSTable components. +RowCacheHitOutOfRange Counter Number of table row cache hits that do not satisfy the query filter, thus went to disk. +RowCacheHit Counter Number of table row cache hits. +RowCacheMiss Counter Number of table row cache misses. +CasPrepare Latency Latency of paxos prepare round. +CasPropose Latency Latency of paxos propose round. +CasCommit Latency Latency of paxos commit round. +PercentRepaired Gauge Percent of table data that is repaired on disk. +BytesRepaired Gauge Size of table data repaired on disk +BytesUnrepaired Gauge Size of table data unrepaired on disk +BytesPendingRepair Gauge Size of table data isolated for an ongoing incremental repair +SpeculativeRetries Counter Number of times speculative retries were sent for this table. +SpeculativeFailedRetries Counter Number of speculative retries that failed to prevent a timeout +SpeculativeInsufficientReplicas Counter Number of speculative retries that couldn't be attempted due to lack of replicas +SpeculativeSampleLatencyNanos Gauge Number of nanoseconds to wait before speculation is attempted. Value may be statically configured or updated periodically based on coordinator latency. +WaitingOnFreeMemtableSpace Histogram Histogram of time spent waiting for free memtable space, either on- or off-heap. +DroppedMutations Counter Number of dropped mutations on this table. +AnticompactionTime Timer Time spent anticompacting before a consistent repair. +ValidationTime Timer Time spent doing validation compaction during repair. +SyncTime Timer Time spent doing streaming during repair. +BytesValidated Histogram Histogram over the amount of bytes read during validation. +PartitionsValidated Histogram Histogram over the number of partitions read during validation. +BytesAnticompacted Counter How many bytes we anticompacted. +BytesMutatedAnticompaction Counter How many bytes we avoided anticompacting because the sstable was fully contained in the repaired range. +MutatedAnticompactionGauge Gauge Ratio of bytes mutated vs total bytes repaired. +ReadRepairRequests Meter Throughput for mutations generated by read-repair. +ShortReadProtectionRequests Meter Throughput for requests to get extra rows during short read protection. +ReplicaFilteringProtectionRequests Meter Throughput for row completion requests during replica filtering protection. +ReplicaFilteringProtectionRowsCachedPerQuery Histogram Histogram of the number of rows cached per query when replica filtering protection is engaged. +============================================ ============== =========== Keyspace Metrics ^^^^^^^^^^^^^^^^ diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index c8af2912b1..21c8e79bfd 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -330,6 +330,8 @@ public class Config public volatile int tombstone_warn_threshold = 1000; public volatile int tombstone_failure_threshold = 100000; + 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; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index a7559f1a0a..397bf024e1 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1863,6 +1863,26 @@ public class DatabaseDescriptor conf.tombstone_failure_threshold = threshold; } + public static int getCachedReplicaRowsWarnThreshold() + { + return conf.replica_filtering_protection.cached_rows_warn_threshold; + } + + public static void setCachedReplicaRowsWarnThreshold(int threshold) + { + conf.replica_filtering_protection.cached_rows_warn_threshold = threshold; + } + + public static int getCachedReplicaRowsFailThreshold() + { + return conf.replica_filtering_protection.cached_rows_fail_threshold; + } + + public static void setCachedReplicaRowsFailThreshold(int threshold) + { + conf.replica_filtering_protection.cached_rows_fail_threshold = threshold; + } + /** * size of commitlog segments to allocate */ diff --git a/src/java/org/apache/cassandra/config/ReplicaFilteringProtectionOptions.java b/src/java/org/apache/cassandra/config/ReplicaFilteringProtectionOptions.java new file mode 100644 index 0000000000..7a755ab068 --- /dev/null +++ b/src/java/org/apache/cassandra/config/ReplicaFilteringProtectionOptions.java @@ -0,0 +1,28 @@ +/* + * 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; + +public class ReplicaFilteringProtectionOptions +{ + public static final int DEFAULT_WARN_THRESHOLD = 2000; + public static final int DEFAULT_FAIL_THRESHOLD = 32000; + + public volatile int cached_rows_warn_threshold = DEFAULT_WARN_THRESHOLD; + public volatile int cached_rows_fail_threshold = DEFAULT_FAIL_THRESHOLD; +} diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java b/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java index 0c4084fdf8..fd69646007 100644 --- a/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java +++ b/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java @@ -95,6 +95,21 @@ public abstract class PartitionIterators } } + /** + * Consumes all rows in the next partition of the provided partition iterator. + */ + public static void consumeNext(PartitionIterator iterator) + { + if (iterator.hasNext()) + { + try (RowIterator partition = iterator.next()) + { + while (partition.hasNext()) + partition.next(); + } + } + } + /** * Wraps the provided iterator so it logs the returned rows for debugging purposes. *

@@ -114,6 +129,38 @@ public abstract class PartitionIterators return Transformation.apply(iterator, new Logger()); } + /** + * Wraps the provided iterator to run a specified action on close. Note that the action will be + * run even if closure of the provided iterator throws an exception. + */ + public static PartitionIterator doOnClose(PartitionIterator delegate, Runnable action) + { + return new PartitionIterator() + { + public void close() + { + try + { + delegate.close(); + } + finally + { + action.run(); + } + } + + public boolean hasNext() + { + return delegate.hasNext(); + } + + public RowIterator next() + { + return delegate.next(); + } + }; + } + private static class SingletonPartitionIterator extends AbstractIterator implements PartitionIterator { private final RowIterator iterator; diff --git a/src/java/org/apache/cassandra/metrics/TableMetrics.java b/src/java/org/apache/cassandra/metrics/TableMetrics.java index bfb261d5a8..24550b069a 100644 --- a/src/java/org/apache/cassandra/metrics/TableMetrics.java +++ b/src/java/org/apache/cassandra/metrics/TableMetrics.java @@ -328,7 +328,16 @@ public class TableMetrics public final Meter readRepairRequests; public final Meter shortReadProtectionRequests; - public final Meter replicaSideFilteringProtectionRequests; + + public final Meter replicaFilteringProtectionRequests; + + /** + * This histogram records the maximum number of rows {@link org.apache.cassandra.service.ReplicaFilteringProtection} + * caches at a point in time per query. With no replica divergence, this is equivalent to the maximum number of + * cached rows in a single partition during a query. It can be helpful when choosing appropriate values for the + * replica_filtering_protection thresholds in cassandra.yaml. + */ + public final Histogram rfpRowsCachedPerQuery; public final EnumMap> samplers; /** @@ -942,7 +951,8 @@ public class TableMetrics readRepairRequests = createTableMeter("ReadRepairRequests"); shortReadProtectionRequests = createTableMeter("ShortReadProtectionRequests"); - replicaSideFilteringProtectionRequests = createTableMeter("ReplicaSideFilteringProtectionRequests"); + replicaFilteringProtectionRequests = createTableMeter("ReplicaFilteringProtectionRequests"); + rfpRowsCachedPerQuery = createHistogram("ReplicaFilteringProtectionRowsCachedPerQuery", true); confirmedRepairedInconsistencies = createTableMeter("RepairedDataInconsistenciesConfirmed", cfs.keyspace.metric.confirmedRepairedInconsistencies); unconfirmedRepairedInconsistencies = createTableMeter("RepairedDataInconsistenciesUnconfirmed", cfs.keyspace.metric.unconfirmedRepairedInconsistencies); @@ -1060,6 +1070,13 @@ public class TableMetrics register(name, alias, tableMeter); return tableMeter; } + + private Histogram createHistogram(String name, boolean considerZeroes) + { + Histogram histogram = Metrics.histogram(factory.createMetricName(name), aliasFactory.createMetricName(name), considerZeroes); + register(name, name, histogram); + return histogram; + } /** * Computes the compression ratio for the specified SSTables diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index a1b3b824d6..0d10418cfa 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -5387,6 +5387,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public void setTombstoneWarnThreshold(int threshold) { DatabaseDescriptor.setTombstoneWarnThreshold(threshold); + logger.info("updated tombstone_warn_threshold to {}", threshold); } public int getTombstoneFailureThreshold() @@ -5397,6 +5398,29 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public void setTombstoneFailureThreshold(int threshold) { DatabaseDescriptor.setTombstoneFailureThreshold(threshold); + logger.info("updated tombstone_failure_threshold to {}", threshold); + } + + public int getCachedReplicaRowsWarnThreshold() + { + return DatabaseDescriptor.getCachedReplicaRowsWarnThreshold(); + } + + public void setCachedReplicaRowsWarnThreshold(int threshold) + { + DatabaseDescriptor.setCachedReplicaRowsWarnThreshold(threshold); + logger.info("updated replica_filtering_protection.cached_rows_warn_threshold to {}", threshold); + } + + public int getCachedReplicaRowsFailThreshold() + { + return DatabaseDescriptor.getCachedReplicaRowsFailThreshold(); + } + + public void setCachedReplicaRowsFailThreshold(int threshold) + { + DatabaseDescriptor.setCachedReplicaRowsFailThreshold(threshold); + logger.info("updated replica_filtering_protection.cached_rows_fail_threshold to {}", threshold); } public int getColumnIndexCacheSize() @@ -5418,7 +5442,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public void setBatchSizeFailureThreshold(int threshold) { DatabaseDescriptor.setBatchSizeFailThresholdInKB(threshold); - logger.info("Updated batch_size_fail_threshold_in_kb to {}", threshold); + logger.info("updated batch_size_fail_threshold_in_kb to {}", threshold); } public int getBatchSizeWarnThreshold() @@ -5470,7 +5494,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public void setHintedHandoffThrottleInKB(int throttleInKB) { DatabaseDescriptor.setHintedHandoffThrottleInKB(throttleInKB); - logger.info("Updated hinted_handoff_throttle_in_kb to {}", throttleInKB); + logger.info("updated hinted_handoff_throttle_in_kb to {}", throttleInKB); } @Override diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 9664769167..3c6bbf9090 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -697,6 +697,18 @@ public interface StorageServiceMBean extends NotificationEmitter /** Sets the threshold for abandoning queries with many tombstones */ public void setTombstoneFailureThreshold(int tombstoneDebugThreshold); + /** Returns the number of rows cached at the coordinator before filtering/index queries log a warning. */ + public int getCachedReplicaRowsWarnThreshold(); + + /** Sets the number of rows cached at the coordinator before filtering/index queries log a warning. */ + public void setCachedReplicaRowsWarnThreshold(int threshold); + + /** Returns the number of rows cached at the coordinator before filtering/index queries fail outright. */ + public int getCachedReplicaRowsFailThreshold(); + + /** Sets the number of rows cached at the coordinator before filtering/index queries fail outright. */ + public void setCachedReplicaRowsFailThreshold(int threshold); + /** Returns the threshold for skipping the column index when caching partition info **/ public int getColumnIndexCacheSize(); /** Sets the threshold for skipping the column index when caching partition info **/ diff --git a/src/java/org/apache/cassandra/service/reads/DataResolver.java b/src/java/org/apache/cassandra/service/reads/DataResolver.java index 30427d0295..eeabb4bbcf 100644 --- a/src/java/org/apache/cassandra/service/reads/DataResolver.java +++ b/src/java/org/apache/cassandra/service/reads/DataResolver.java @@ -25,6 +25,7 @@ import java.util.function.UnaryOperator; import com.google.common.base.Joiner; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.statements.schema.IndexTarget; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.DeletionTime; @@ -171,6 +172,7 @@ public class DataResolver, P extends ReplicaPlan.ForRead< return context.needShortReadProtection() ? ShortReadProtection.extend(context.replicas.get(i), + () -> responses.clearUnsafe(i), originalResponse, command, context.mergedResultCounter, @@ -199,20 +201,20 @@ public class DataResolver, P extends ReplicaPlan.ForRead< { // Protecting against inconsistent replica filtering (some replica returning a row that is outdated but that // wouldn't be removed by normal reconciliation because up-to-date replica have filtered the up-to-date version - // of that row) works in 3 steps: - // 1) we read the full response just to collect rows that may be outdated (the ones we got from some - // replica but didn't got any response for other; it could be those other replica have filtered a more - // up-to-date result). In doing so, we do not count any of such "potentially outdated" row towards the - // query limit. This simulate the worst case scenario where all those "potentially outdated" rows are - // indeed outdated, and thus make sure we are guaranteed to read enough results (thanks to short read - // protection). - // 2) we query all the replica/rows we need to rule out whether those "potentially outdated" rows are outdated - // or not. - // 3) we re-read cached copies of each replica response using the "normal" read path merge with read-repair, - // but where for each replica we use their original response _plus_ the additional rows queried in the - // previous step (and apply the command#rowFilter() on the full result). Since the first phase has - // pessimistically collected enough results for the case where all potentially outdated results are indeed - // outdated, we shouldn't need further short-read protection requests during this phase. + // of that row) involves 3 main elements: + // 1) We combine short-read protection and a merge listener that identifies potentially "out-of-date" + // rows to create an iterator that is guaranteed to produce enough valid row results to satisfy the query + // limit if enough actually exist. A row is considered out-of-date if its merged from is non-empty and we + // receive not response from at least one replica. In this case, it is possible that filtering at the + // "silent" replica has produced a more up-to-date result. + // 2) This iterator is passed to the standard resolution process with read-repair, but is first wrapped in a + // response provider that lazily "completes" potentially out-of-date rows by directly querying them on the + // replicas that were previously silent. As this iterator is consumed, it caches valid data for potentially + // out-of-date rows, and this cached data is merged with the fetched data as rows are requested. If there + // is no replica divergence, only rows in the partition being evalutated will be cached (then released + // when the partition is consumed). + // 3) After a "complete" row is materialized, it must pass the row filter supplied by the original query + // before it counts against the limit. // We need separate contexts, as each context has his own counter ResolveContext firstPhaseContext = new ResolveContext(replicas); @@ -221,26 +223,22 @@ public class DataResolver, P extends ReplicaPlan.ForRead< command, replicaPlan().consistencyLevel(), queryStartNanoTime, - firstPhaseContext.replicas); + firstPhaseContext.replicas, + DatabaseDescriptor.getCachedReplicaRowsWarnThreshold(), + DatabaseDescriptor.getCachedReplicaRowsFailThreshold()); + PartitionIterator firstPhasePartitions = resolveInternal(firstPhaseContext, rfp.mergeController(), i -> shortReadProtectedResponse(i, firstPhaseContext), UnaryOperator.identity()); - // Consume the first phase partitions to populate the replica filtering protection with both those materialized - // partitions and the primary keys to be fetched. - PartitionIterators.consume(firstPhasePartitions); - firstPhasePartitions.close(); + PartitionIterator completedPartitions = resolveWithReadRepair(secondPhaseContext, + i -> rfp.queryProtectedPartitions(firstPhasePartitions, i), + results -> command.rowFilter().filter(results, command.metadata(), command.nowInSec()), + repairedDataTracker); - // After reading the entire query results the protection helper should have cached all the partitions so we can - // clear the responses accumulator for the sake of memory usage, given that the second phase might take long if - // it needs to query replicas. - responses.clearUnsafe(); - - return resolveWithReadRepair(secondPhaseContext, - rfp::queryProtectedPartitions, - results -> command.rowFilter().filter(results, command.metadata(), command.nowInSec()), - repairedDataTracker); + // Ensure that the RFP instance has a chance to record metrics when the iterator closes. + return PartitionIterators.doOnClose(completedPartitions, firstPhasePartitions::close); } @SuppressWarnings("resource") @@ -269,7 +267,8 @@ public class DataResolver, P extends ReplicaPlan.ForRead< */ UnfilteredPartitionIterator merged = UnfilteredPartitionIterators.merge(results, mergeListener); - FilteredPartitions filtered = FilteredPartitions.filter(merged, new Filter(command.nowInSec(), command.metadata().enforceStrictLiveness())); + Filter filter = new Filter(command.nowInSec(), command.metadata().enforceStrictLiveness()); + FilteredPartitions filtered = FilteredPartitions.filter(merged, filter); PartitionIterator counted = Transformation.apply(preCountFilter.apply(filtered), context.mergedResultCounter); return Transformation.apply(counted, new EmptyPartitionsDiscarder()); } diff --git a/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java b/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java index b4a3cb5c2b..4f0ad4d06b 100644 --- a/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java +++ b/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java @@ -18,14 +18,14 @@ package org.apache.cassandra.service.reads; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; -import java.util.Iterator; import java.util.List; import java.util.NavigableSet; -import java.util.SortedMap; -import java.util.TreeMap; -import java.util.stream.Collectors; +import java.util.concurrent.TimeUnit; +import java.util.Queue; +import java.util.function.Function; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -45,6 +45,8 @@ import org.apache.cassandra.db.filter.ClusteringIndexFilter; import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter; import org.apache.cassandra.db.filter.DataLimits; import org.apache.cassandra.db.filter.RowFilter; +import org.apache.cassandra.db.partitions.PartitionIterator; +import org.apache.cassandra.db.partitions.PartitionIterators; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; import org.apache.cassandra.db.rows.EncodingStats; @@ -54,18 +56,22 @@ import org.apache.cassandra.db.rows.Rows; import org.apache.cassandra.db.rows.Unfiltered; import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.db.rows.UnfilteredRowIterators; +import org.apache.cassandra.exceptions.OverloadedException; import org.apache.cassandra.exceptions.ReadTimeoutException; import org.apache.cassandra.exceptions.UnavailableException; import org.apache.cassandra.locator.Endpoints; +import org.apache.cassandra.locator.EndpointsForToken; import org.apache.cassandra.locator.Replica; import org.apache.cassandra.locator.ReplicaPlan; import org.apache.cassandra.locator.ReplicaPlans; import org.apache.cassandra.metrics.TableMetrics; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.service.ClientWarn; import org.apache.cassandra.service.StorageProxy; import org.apache.cassandra.service.reads.repair.NoopReadRepair; import org.apache.cassandra.tracing.Tracing; +import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.btree.BTreeSet; /** @@ -77,11 +83,15 @@ import org.apache.cassandra.utils.btree.BTreeSet; * the rows in a replica response that don't have a corresponding row in other replica responses, and requests them by * primary key to the "silent" replicas in a second fetch round. *

- * See CASSANDRA-8272 and CASSANDRA-8273 for further details. + * See CASSANDRA-8272, CASSANDRA-8273, and CASSANDRA-15907 for further details. */ class ReplicaFilteringProtection> { private static final Logger logger = LoggerFactory.getLogger(ReplicaFilteringProtection.class); + private static final NoSpamLogger oneMinuteLogger = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES); + + private static final Function NULL_TO_NO_STATS = + rowIterator -> rowIterator == null ? EncodingStats.NO_STATS : rowIterator.stats(); private final Keyspace keyspace; private final ReadCommand command; @@ -90,117 +100,53 @@ class ReplicaFilteringProtection> private final E sources; private final TableMetrics tableMetrics; - /** - * Per-source primary keys of the rows that might be outdated so they need to be fetched. - * For outdated static rows we use an empty builder to signal it has to be queried. - */ - private final List>> rowsToFetch; + private final int cachedRowsWarnThreshold; + private final int cachedRowsFailThreshold; + + /** Tracks whether or not we've already hit the warning threshold while evaluating a partition. */ + private boolean hitWarningThreshold = false; + + private int currentRowsCached = 0; // tracks the current number of cached rows + private int maxRowsCached = 0; // tracks the high watermark for the number of cached rows /** - * Per-source list of all the partitions seen by the merge listener, to be merged with the extra fetched rows. + * Per-source list of the pending partitions seen by the merge listener, to be merged with the extra fetched rows. */ - private final List> originalPartitions; + private final List> originalPartitions; ReplicaFilteringProtection(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistency, long queryStartNanoTime, - E sources) + E sources, + int cachedRowsWarnThreshold, + int cachedRowsFailThreshold) { this.keyspace = keyspace; this.command = command; this.consistency = consistency; this.queryStartNanoTime = queryStartNanoTime; this.sources = sources; - this.rowsToFetch = new ArrayList<>(sources.size()); this.originalPartitions = new ArrayList<>(sources.size()); - for (Replica ignored : sources) + for (int i = 0; i < sources.size(); i++) { - rowsToFetch.add(new TreeMap<>()); - originalPartitions.add(new ArrayList<>()); + originalPartitions.add(new ArrayDeque<>()); } tableMetrics = ColumnFamilyStore.metricsFor(command.metadata().id); + + this.cachedRowsWarnThreshold = cachedRowsWarnThreshold; + this.cachedRowsFailThreshold = cachedRowsFailThreshold; } - private BTreeSet.Builder getOrCreateToFetch(int source, DecoratedKey partitionKey) + private UnfilteredPartitionIterator executeReadCommand(ReadCommand cmd, Replica source, ReplicaPlan.Shared replicaPlan) { - return rowsToFetch.get(source).computeIfAbsent(partitionKey, k -> BTreeSet.builder(command.metadata().comparator)); - } + @SuppressWarnings("unchecked") + DataResolver resolver = + new DataResolver<>(cmd, replicaPlan, (NoopReadRepair) NoopReadRepair.instance, queryStartNanoTime); - /** - * Returns the protected results for the specified replica. These are generated fetching the extra rows and merging - * them with the cached original filtered results for that replica. - * - * @param source the source - * @return the protected results for the specified replica - */ - UnfilteredPartitionIterator queryProtectedPartitions(int source) - { - UnfilteredPartitionIterator original = makeIterator(originalPartitions.get(source)); - SortedMap> toFetch = rowsToFetch.get(source); - - if (toFetch.isEmpty()) - return original; - - // TODO: this would be more efficient if we had multi-key queries internally - List fetched = toFetch.keySet() - .stream() - .map(k -> querySourceOnKey(source, k)) - .collect(Collectors.toList()); - - return UnfilteredPartitionIterators.merge(Arrays.asList(original, UnfilteredPartitionIterators.concat(fetched)), null); - } - - private UnfilteredPartitionIterator querySourceOnKey(int i, DecoratedKey key) - { - BTreeSet.Builder builder = rowsToFetch.get(i).get(key); - assert builder != null; // We're calling this on the result of rowsToFetch.get(i).keySet() - - Replica source = sources.get(i); - NavigableSet clusterings = builder.build(); - tableMetrics.replicaSideFilteringProtectionRequests.mark(); - if (logger.isTraceEnabled()) - logger.trace("Requesting rows {} in partition {} from {} for replica-side filtering protection", - clusterings, key, source); - Tracing.trace("Requesting {} rows in partition {} from {} for replica-side filtering protection", - clusterings.size(), key, source); - - // build the read command taking into account that we could be requesting only in the static row - DataLimits limits = clusterings.isEmpty() ? DataLimits.cqlLimits(1) : DataLimits.NONE; - ClusteringIndexFilter filter = new ClusteringIndexNamesFilter(clusterings, command.isReversed()); - SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(command.metadata(), - command.nowInSec(), - command.columnFilter(), - RowFilter.NONE, - limits, - key, - filter); - - ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forSingleReplicaRead(keyspace, key.getToken(), source); - ReplicaPlan.SharedForTokenRead sharedReplicaPlan = ReplicaPlan.shared(replicaPlan); - try - { - return executeReadCommand(cmd, source, sharedReplicaPlan); - } - catch (ReadTimeoutException e) - { - int blockFor = consistency.blockFor(keyspace); - throw new ReadTimeoutException(consistency, blockFor - 1, blockFor, true); - } - catch (UnavailableException e) - { - int blockFor = consistency.blockFor(keyspace); - throw UnavailableException.create(consistency, blockFor, blockFor - 1); - } - } - - private , P extends ReplicaPlan.ForRead> - UnfilteredPartitionIterator executeReadCommand(ReadCommand cmd, Replica source, ReplicaPlan.Shared replicaPlan) - { - DataResolver resolver = new DataResolver<>(cmd, replicaPlan, (NoopReadRepair)NoopReadRepair.instance, queryStartNanoTime); - ReadCallback handler = new ReadCallback<>(resolver, cmd, replicaPlan, queryStartNanoTime); + ReadCallback handler = new ReadCallback<>(resolver, cmd, replicaPlan, queryStartNanoTime); if (source.isSelf()) { @@ -227,83 +173,124 @@ class ReplicaFilteringProtection> *

* The listener will track both the accepted data and the primary keys of the rows that are considered as outdated. * That way, once the query results would have been merged using this listener, further calls to - * {@link #queryProtectedPartitions(int)} will use the collected data to return a copy of the + * {@link #queryProtectedPartitions(PartitionIterator, int)} will use the collected data to return a copy of the * data originally collected from the specified replica, completed with the potentially outdated rows. */ UnfilteredPartitionIterators.MergeListener mergeController() { - return (partitionKey, versions) -> { - - PartitionBuilder[] builders = new PartitionBuilder[sources.size()]; - - for (int i = 0; i < sources.size(); i++) - builders[i] = new PartitionBuilder(command, partitionKey, columns(versions), stats(versions)); - - return new UnfilteredRowIterators.MergeListener() + return new UnfilteredPartitionIterators.MergeListener() + { + @Override + public void close() { - @Override - public void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions) + // If we hit the failure threshold before consuming a single partition, record the current rows cached. + tableMetrics.rfpRowsCachedPerQuery.update(Math.max(currentRowsCached, maxRowsCached)); + } + + @Override + public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List versions) + { + List builders = new ArrayList<>(sources.size()); + RegularAndStaticColumns columns = columns(versions); + EncodingStats stats = EncodingStats.merge(versions, NULL_TO_NO_STATS); + + for (int i = 0; i < sources.size(); i++) + builders.add(i, new PartitionBuilder(partitionKey, sources.get(i), columns, stats)); + + return new UnfilteredRowIterators.MergeListener() { - // cache the deletion time versions to be able to regenerate the original row iterator - for (int i = 0; i < versions.length; i++) - builders[i].setDeletionTime(versions[i]); - } - - @Override - public Row onMergedRows(Row merged, Row[] versions) - { - // cache the row versions to be able to regenerate the original row iterator - for (int i = 0; i < versions.length; i++) - builders[i].addRow(versions[i]); - - if (merged.isEmpty()) - return merged; - - boolean isPotentiallyOutdated = false; - boolean isStatic = merged.isStatic(); - for (int i = 0; i < versions.length; i++) + @Override + public void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions) { - Row version = versions[i]; - if (version == null || (isStatic && version.isEmpty())) - { - isPotentiallyOutdated = true; - BTreeSet.Builder toFetch = getOrCreateToFetch(i, partitionKey); - // Note that for static, we shouldn't add the clustering to the clustering set (the - // ClusteringIndexNamesFilter we'll build from this later does not expect it), but the fact - // we created a builder in the first place will act as a marker that the static row must be - // fetched, even if no other rows are added for this partition. - if (!isStatic) - toFetch.add(merged.clustering()); - } + // cache the deletion time versions to be able to regenerate the original row iterator + for (int i = 0; i < versions.length; i++) + builders.get(i).setDeletionTime(versions[i]); } - // If the row is potentially outdated (because some replica didn't send anything and so it _may_ be - // an outdated result that is only present because other replica have filtered the up-to-date result - // out), then we skip the row. In other words, the results of the initial merging of results by this - // protection assume the worst case scenario where every row that might be outdated actually is. - // This ensures that during this first phase (collecting additional row to fetch) we are guaranteed - // to look at enough data to ultimately fulfill the query limit. - return isPotentiallyOutdated ? null : merged; - } + @Override + public Row onMergedRows(Row merged, Row[] versions) + { + // cache the row versions to be able to regenerate the original row iterator + for (int i = 0; i < versions.length; i++) + builders.get(i).addRow(versions[i]); - @Override - public void onMergedRangeTombstoneMarkers(RangeTombstoneMarker merged, RangeTombstoneMarker[] versions) - { - // cache the marker versions to be able to regenerate the original row iterator - for (int i = 0; i < versions.length; i++) - builders[i].addRangeTombstoneMarker(versions[i]); - } + if (merged.isEmpty()) + return merged; - @Override - public void close() - { - for (int i = 0; i < sources.size(); i++) - originalPartitions.get(i).add(builders[i]); - } - }; + boolean isPotentiallyOutdated = false; + boolean isStatic = merged.isStatic(); + for (int i = 0; i < versions.length; i++) + { + Row version = versions[i]; + if (version == null || (isStatic && version.isEmpty())) + { + isPotentiallyOutdated = true; + builders.get(i).addToFetch(merged); + } + } + + // If the row is potentially outdated (because some replica didn't send anything and so it _may_ be + // an outdated result that is only present because other replica have filtered the up-to-date result + // out), then we skip the row. In other words, the results of the initial merging of results by this + // protection assume the worst case scenario where every row that might be outdated actually is. + // This ensures that during this first phase (collecting additional row to fetch) we are guaranteed + // to look at enough data to ultimately fulfill the query limit. + return isPotentiallyOutdated ? null : merged; + } + + @Override + public void onMergedRangeTombstoneMarkers(RangeTombstoneMarker merged, RangeTombstoneMarker[] versions) + { + // cache the marker versions to be able to regenerate the original row iterator + for (int i = 0; i < versions.length; i++) + builders.get(i).addRangeTombstoneMarker(versions[i]); + } + + @Override + public void close() + { + for (int i = 0; i < sources.size(); i++) + originalPartitions.get(i).add(builders.get(i)); + } + }; + } }; } + private void incrementCachedRows() + { + currentRowsCached++; + + if (currentRowsCached == cachedRowsFailThreshold + 1) + { + String message = String.format("Replica filtering protection has cached over %d rows during query %s. " + + "(See 'cached_replica_rows_fail_threshold' in cassandra.yaml.)", + cachedRowsFailThreshold, command.toCQLString()); + + logger.error(message); + Tracing.trace(message); + throw new OverloadedException(message); + } + else if (currentRowsCached == cachedRowsWarnThreshold + 1 && !hitWarningThreshold) + { + hitWarningThreshold = true; + + String message = String.format("Replica filtering protection has cached over %d rows during query %s. " + + "(See 'cached_replica_rows_warn_threshold' in cassandra.yaml.)", + cachedRowsWarnThreshold, command.toCQLString()); + + ClientWarn.instance.warn(message); + oneMinuteLogger.warn(message); + Tracing.trace(message); + } + } + + private void releaseCachedRows(int count) + { + maxRowsCached = Math.max(maxRowsCached, currentRowsCached); + currentRowsCached -= count; + } + private static RegularAndStaticColumns columns(List versions) { Columns statics = Columns.NONE; @@ -320,24 +307,19 @@ class ReplicaFilteringProtection> return new RegularAndStaticColumns(statics, regulars); } - private static EncodingStats stats(List iterators) - { - EncodingStats stats = EncodingStats.NO_STATS; - for (UnfilteredRowIterator iter : iterators) - { - if (iter == null) - continue; - - stats = stats.mergeWith(iter.stats()); - } - return stats; - } - - private UnfilteredPartitionIterator makeIterator(List builders) + /** + * Returns the protected results for the specified replica. These are generated fetching the extra rows and merging + * them with the cached original filtered results for that replica. + * + * @param merged the first iteration partitions, that should have been read used with the {@link #mergeController()} + * @param source the source + * @return the protected results for the specified replica + */ + UnfilteredPartitionIterator queryProtectedPartitions(PartitionIterator merged, int source) { return new UnfilteredPartitionIterator() { - final Iterator iterator = builders.iterator(); + final Queue partitions = originalPartitions.get(source); @Override public TableMetadata metadata() @@ -346,43 +328,49 @@ class ReplicaFilteringProtection> } @Override - public void close() - { - // nothing to do here - } + public void close() { } @Override public boolean hasNext() { - return iterator.hasNext(); + // If there are no cached partition builders for this source, advance the first phase iterator, which + // will force the RFP merge listener to load at least the next protected partition. Note that this may + // load more than one partition if any divergence between replicas is discovered by the merge listener. + if (partitions.isEmpty()) + { + PartitionIterators.consumeNext(merged); + } + + return !partitions.isEmpty(); } @Override public UnfilteredRowIterator next() { - return iterator.next().build(); + PartitionBuilder builder = partitions.poll(); + assert builder != null; + return builder.protectedPartition(); } }; } - private static class PartitionBuilder + private class PartitionBuilder { - private final ReadCommand command; - private final DecoratedKey partitionKey; + private final DecoratedKey key; + private final Replica source; private final RegularAndStaticColumns columns; private final EncodingStats stats; private DeletionTime deletionTime; private Row staticRow = Rows.EMPTY_STATIC_ROW; - private final List contents = new ArrayList<>(); + private final Queue contents = new ArrayDeque<>(); + private BTreeSet.Builder toFetch; + private int partitionRowsCached; - private PartitionBuilder(ReadCommand command, - DecoratedKey partitionKey, - RegularAndStaticColumns columns, - EncodingStats stats) + private PartitionBuilder(DecoratedKey key, Replica source, RegularAndStaticColumns columns, EncodingStats stats) { - this.command = command; - this.partitionKey = partitionKey; + this.key = key; + this.source = source; this.columns = columns; this.stats = stats; } @@ -394,6 +382,12 @@ class ReplicaFilteringProtection> private void addRow(Row row) { + partitionRowsCached++; + + incrementCachedRows(); + + // Note that even null rows are counted against the row caching limit. The assumption is that + // a subsequent protection query will later fetch the row onto the heap anyway. if (row == null) return; @@ -409,12 +403,23 @@ class ReplicaFilteringProtection> contents.add(marker); } - private UnfilteredRowIterator build() + private void addToFetch(Row row) + { + if (toFetch == null) + toFetch = BTreeSet.builder(command.metadata().comparator); + + // Note that for static, we shouldn't add the clustering to the clustering set (the + // ClusteringIndexNamesFilter we'll build from this later does not expect it), but the fact + // we created a builder in the first place will act as a marker that the static row must be + // fetched, even if no other rows are added for this partition. + if (!row.isStatic()) + toFetch.add(row.clustering()); + } + + private UnfilteredRowIterator originalPartition() { return new UnfilteredRowIterator() { - final Iterator iterator = contents.iterator(); - @Override public DeletionTime partitionLevelDeletion() { @@ -448,7 +453,7 @@ class ReplicaFilteringProtection> @Override public DecoratedKey partitionKey() { - return partitionKey; + return key; } @Override @@ -460,21 +465,86 @@ class ReplicaFilteringProtection> @Override public void close() { - // nothing to do here + releaseCachedRows(partitionRowsCached); } @Override public boolean hasNext() { - return iterator.hasNext(); + return !contents.isEmpty(); } @Override public Unfiltered next() { - return iterator.next(); + return contents.poll(); } }; } + + private UnfilteredRowIterator protectedPartition() + { + UnfilteredRowIterator original = originalPartition(); + + if (toFetch != null) + { + try (UnfilteredPartitionIterator partitions = fetchFromSource()) + { + if (partitions.hasNext()) + { + try (UnfilteredRowIterator fetchedRows = partitions.next()) + { + return UnfilteredRowIterators.merge(Arrays.asList(original, fetchedRows)); + } + } + } + } + + return original; + } + + private UnfilteredPartitionIterator fetchFromSource() + { + assert toFetch != null; + + NavigableSet clusterings = toFetch.build(); + tableMetrics.replicaFilteringProtectionRequests.mark(); + + if (logger.isTraceEnabled()) + logger.trace("Requesting rows {} in partition {} from {} for replica filtering protection", + clusterings, key, source); + + Tracing.trace("Requesting {} rows in partition {} from {} for replica filtering protection", + clusterings.size(), key, source); + + // build the read command taking into account that we could be requesting only in the static row + DataLimits limits = clusterings.isEmpty() ? DataLimits.cqlLimits(1) : DataLimits.NONE; + ClusteringIndexFilter filter = new ClusteringIndexNamesFilter(clusterings, command.isReversed()); + SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(command.metadata(), + command.nowInSec(), + command.columnFilter(), + RowFilter.NONE, + limits, + key, + filter); + + ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forSingleReplicaRead(keyspace, key.getToken(), source); + ReplicaPlan.SharedForTokenRead sharedReplicaPlan = ReplicaPlan.shared(replicaPlan); + + try + { + return executeReadCommand(cmd, source, sharedReplicaPlan); + } + catch (ReadTimeoutException e) + { + int blockFor = consistency.blockFor(keyspace); + throw new ReadTimeoutException(consistency, blockFor - 1, blockFor, true); + } + catch (UnavailableException e) + { + int blockFor = consistency.blockFor(keyspace); + throw UnavailableException.create(consistency, blockFor, blockFor - 1); + } + } } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java b/src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java index 59edc5a0f1..42676b6a77 100644 --- a/src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java +++ b/src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java @@ -53,6 +53,8 @@ public class ShortReadPartitionsProtection extends Transformation, P extends ReplicaPlan.ForRead> - implements ReadRepair +public class NoopReadRepair, P extends ReplicaPlan.ForRead> implements ReadRepair { public static final NoopReadRepair instance = new NoopReadRepair(); diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index d43a0154fc..17e67a20c7 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -299,7 +299,7 @@ public abstract class Message return tracingId; } - Message setWarnings(List warnings) + public Message setWarnings(List warnings) { this.warnings = warnings; return this; diff --git a/src/java/org/apache/cassandra/utils/concurrent/Accumulator.java b/src/java/org/apache/cassandra/utils/concurrent/Accumulator.java index 7adb33b2b5..3ebbdce544 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Accumulator.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Accumulator.java @@ -20,7 +20,6 @@ package org.apache.cassandra.utils.concurrent; import java.util.AbstractCollection; import java.util.Collection; -import java.util.Arrays; import java.util.Iterator; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; @@ -159,14 +158,12 @@ public class Accumulator } /** - * Removes all of the elements from this accumulator. + * Removes element at the speficied index from this accumulator. * * This method is not thread-safe when used concurrently with {@link #add(Object)}. */ - public void clearUnsafe() + public void clearUnsafe(int i) { - nextIndexUpdater.set(this, 0); - presentCountUpdater.set(this, 0); - Arrays.fill(values, null); + values[i] = null; } } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java index 2ee209d91d..0b6fe6e378 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java @@ -39,6 +39,7 @@ import org.apache.cassandra.distributed.api.QueryResult; import org.apache.cassandra.distributed.api.QueryResults; import org.apache.cassandra.distributed.api.SimpleQueryResult; import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.ClientWarn; import org.apache.cassandra.service.QueryState; import org.apache.cassandra.service.pager.QueryPager; import org.apache.cassandra.transport.ProtocolVersion; @@ -91,6 +92,11 @@ public class Coordinator implements ICoordinator boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue)); prepared.validate(QueryState.forInternalCalls().getClientState()); + + // Start capturing warnings on this thread. Note that this will implicitly clear out any previous + // warnings as it sets a new State instance on the ThreadLocal. + ClientWarn.instance.captureWarnings(); + ResultMessage res = prepared.execute(QueryState.forInternalCalls(), QueryOptions.create(toCassandraCL(consistencyLevel), boundBBValues, @@ -102,6 +108,10 @@ public class Coordinator implements ICoordinator null), System.nanoTime()); + // Collect warnings reported during the query. + if (res != null) + res.setWarnings(ClientWarn.instance.getWarnings()); + return RowUtil.toQueryResult(res); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java index 50d501ec49..876ce297a9 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java @@ -19,6 +19,7 @@ package org.apache.cassandra.distributed.impl; import java.nio.ByteBuffer; +import java.util.Collections; import java.util.Iterator; import java.util.List; @@ -41,7 +42,11 @@ public class RowUtil ResultMessage.Rows rows = (ResultMessage.Rows) res; String[] names = getColumnNames(rows.result.metadata.names); Object[][] results = RowUtil.toObjects(rows); - return new SimpleQueryResult(names, results); + + // Warnings may be null here, due to ClientWarn#getWarnings() handling of empty warning lists. + List warnings = res.getWarnings(); + + return new SimpleQueryResult(names, results, warnings == null ? Collections.emptyList() : warnings); } else { diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java new file mode 100644 index 0000000000..f891dfee5c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java @@ -0,0 +1,244 @@ +/* + * 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 java.io.IOException; +import java.util.List; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.SimpleQueryResult; +import org.apache.cassandra.exceptions.OverloadedException; +import org.apache.cassandra.service.StorageService; + +import static org.apache.cassandra.config.ReplicaFilteringProtectionOptions.DEFAULT_FAIL_THRESHOLD; +import static org.apache.cassandra.config.ReplicaFilteringProtectionOptions.DEFAULT_WARN_THRESHOLD; +import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; +import static org.junit.Assert.assertEquals; + +/** + * Exercises the functionality of {@link org.apache.cassandra.service.reads.ReplicaFilteringProtection}, the + * mechanism that ensures distributed index and filtering queries at read consistency levels > ONE/LOCAL_ONE + * avoid stale replica results. + */ +public class ReplicaFilteringProtectionTest extends TestBaseImpl +{ + private static final int REPLICAS = 2; + private static final int ROWS = 3; + + private static Cluster cluster; + + @BeforeClass + public static void setup() throws IOException + { + cluster = init(Cluster.build() + .withNodes(REPLICAS) + .withConfig(config -> config.set("hinted_handoff_enabled", false) + .set("commitlog_sync", "batch") + .set("num_tokens", 1)).start()); + + // Make sure we start w/ the correct defaults: + cluster.get(1).runOnInstance(() -> assertEquals(DEFAULT_WARN_THRESHOLD, StorageService.instance.getCachedReplicaRowsWarnThreshold())); + cluster.get(1).runOnInstance(() -> assertEquals(DEFAULT_FAIL_THRESHOLD, StorageService.instance.getCachedReplicaRowsFailThreshold())); + } + + @AfterClass + public static void teardown() + { + cluster.close(); + } + + @Test + public void testMissedUpdatesBelowCachingWarnThreshold() + { + String tableName = "missed_updates_no_warning"; + cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)")); + + // The warning threshold provided is one more than the total number of rows returned + // to the coordinator from all replicas and therefore should not be triggered. + testMissedUpdates(tableName, REPLICAS * ROWS, Integer.MAX_VALUE, false); + } + + @Test + public void testMissedUpdatesAboveCachingWarnThreshold() + { + String tableName = "missed_updates_cache_warn"; + cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)")); + + // The warning threshold provided is one less than the total number of rows returned + // to the coordinator from all replicas and therefore should be triggered but not fail the query. + testMissedUpdates(tableName, REPLICAS * ROWS - 1, Integer.MAX_VALUE, true); + } + + @Test + public void testMissedUpdatesAroundCachingFailThreshold() + { + String tableName = "missed_updates_cache_fail"; + cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)")); + + // The failure threshold provided is exactly the total number of rows returned + // to the coordinator from all replicas and therefore should just warn. + testMissedUpdates(tableName, 1, REPLICAS * ROWS, true); + + try + { + // The failure threshold provided is one less than the total number of rows returned + // to the coordinator from all replicas and therefore should fail the query. + testMissedUpdates(tableName, 1, REPLICAS * ROWS - 1, true); + } + catch (RuntimeException e) + { + assertEquals(e.getClass().getName(), OverloadedException.class.getName()); + } + } + + private void testMissedUpdates(String tableName, int warnThreshold, int failThreshold, boolean shouldWarn) + { + cluster.get(1).runOnInstance(() -> StorageService.instance.setCachedReplicaRowsWarnThreshold(warnThreshold)); + cluster.get(1).runOnInstance(() -> StorageService.instance.setCachedReplicaRowsFailThreshold(failThreshold)); + + String fullTableName = KEYSPACE + '.' + tableName; + + // Case 1: Insert and query rows at ALL to verify base line. + for (int i = 0; i < ROWS; i++) + { + cluster.coordinator(1).execute("INSERT INTO " + fullTableName + "(k, v) VALUES (?, 'old')", ALL, i); + } + + long histogramSampleCount = rowsCachedPerQueryCount(cluster.get(1), tableName); + + String query = "SELECT * FROM " + fullTableName + " WHERE v = ? LIMIT ? ALLOW FILTERING"; + + Object[][] initialRows = cluster.coordinator(1).execute(query, ALL, "old", ROWS); + assertRows(initialRows, row(1, "old"), row(0, "old"), row(2, "old")); + + // Make sure only one sample was recorded for the query. + assertEquals(histogramSampleCount + 1, rowsCachedPerQueryCount(cluster.get(1), tableName)); + + // Case 2: Update all rows on only one replica, leaving the entire dataset of the remaining replica out-of-date. + updateAllRowsOn(1, fullTableName, "new"); + + // The replica that missed the results creates a mismatch at every row, and we therefore cache a version + // of that row for all replicas. + SimpleQueryResult oldResult = cluster.coordinator(1).executeWithResult(query, ALL, "old", ROWS); + assertRows(oldResult.toObjectArrays()); + verifyWarningState(shouldWarn, oldResult); + + // We should have made 3 row "completion" requests. + assertEquals(ROWS, protectionQueryCount(cluster.get(1), tableName)); + + // In all cases above, the queries should be caching 1 row per partition per replica, but + // 6 for the whole query, given every row is potentially stale. + assertEquals(ROWS * REPLICAS, maxRowsCachedPerQuery(cluster.get(1), tableName)); + + // Make sure only one more sample was recorded for the query. + assertEquals(histogramSampleCount + 2, rowsCachedPerQueryCount(cluster.get(1), tableName)); + + // Case 3: Observe the effects of blocking read-repair. + + // The previous query peforms a blocking read-repair, which removes replica divergence. This + // will only warn, therefore, if the warning threshold is actually below the number of replicas. + // (i.e. The row cache counter is decremented/reset as each partition is consumed.) + SimpleQueryResult newResult = cluster.coordinator(1).executeWithResult(query, ALL, "new", ROWS); + Object[][] newRows = newResult.toObjectArrays(); + assertRows(newRows, row(1, "new"), row(0, "new"), row(2, "new")); + + verifyWarningState(warnThreshold < REPLICAS, newResult); + + // We still sould only have made 3 row "completion" requests, with no replica divergence in the last query. + assertEquals(ROWS, protectionQueryCount(cluster.get(1), tableName)); + + // With no replica divergence, we only cache a single partition at a time across 2 replicas. + assertEquals(REPLICAS, minRowsCachedPerQuery(cluster.get(1), tableName)); + + // Make sure only one more sample was recorded for the query. + assertEquals(histogramSampleCount + 3, rowsCachedPerQueryCount(cluster.get(1), tableName)); + + // Case 4: Introduce another mismatch by updating all rows on only one replica. + + updateAllRowsOn(1, fullTableName, "future"); + + // Another mismatch is introduced, and we once again cache a version of each row during resolution. + SimpleQueryResult futureResult = cluster.coordinator(1).executeWithResult(query, ALL, "future", ROWS); + Object[][] futureRows = futureResult.toObjectArrays(); + assertRows(futureRows, row(1, "future"), row(0, "future"), row(2, "future")); + + verifyWarningState(shouldWarn, futureResult); + + // We sould have made 3 more row "completion" requests. + assertEquals(ROWS * 2, protectionQueryCount(cluster.get(1), tableName)); + + // In all cases above, the queries should be caching 1 row per partition, but 6 for the + // whole query, given every row is potentially stale. + assertEquals(ROWS * REPLICAS, maxRowsCachedPerQuery(cluster.get(1), tableName)); + + // Make sure only one more sample was recorded for the query. + assertEquals(histogramSampleCount + 4, rowsCachedPerQueryCount(cluster.get(1), tableName)); + } + + private void updateAllRowsOn(int node, String table, String value) + { + for (int i = 0; i < ROWS; i++) + { + cluster.get(node).executeInternal("UPDATE " + table + " SET v = ? WHERE k = ?", value, i); + } + } + + private void verifyWarningState(boolean shouldWarn, SimpleQueryResult futureResult) + { + List futureWarnings = futureResult.warnings(); + assertEquals(shouldWarn, futureWarnings.stream().anyMatch(w -> w.contains("cached_replica_rows_warn_threshold"))); + assertEquals(shouldWarn ? 1 : 0, futureWarnings.size()); + } + + private long protectionQueryCount(IInvokableInstance instance, String tableName) + { + return instance.callOnInstance(() -> Keyspace.open(KEYSPACE) + .getColumnFamilyStore(tableName) + .metric.replicaFilteringProtectionRequests.getCount()); + } + + private long maxRowsCachedPerQuery(IInvokableInstance instance, String tableName) + { + return instance.callOnInstance(() -> Keyspace.open(KEYSPACE) + .getColumnFamilyStore(tableName) + .metric.rfpRowsCachedPerQuery.getSnapshot().getMax()); + } + + private long minRowsCachedPerQuery(IInvokableInstance instance, String tableName) + { + return instance.callOnInstance(() -> Keyspace.open(KEYSPACE) + .getColumnFamilyStore(tableName) + .metric.rfpRowsCachedPerQuery.getSnapshot().getMin()); + } + + private long rowsCachedPerQueryCount(IInvokableInstance instance, String tableName) + { + return instance.callOnInstance(() -> Keyspace.open(KEYSPACE) + .getColumnFamilyStore(tableName) + .metric.rfpRowsCachedPerQuery.getCount()); + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/utils/concurrent/AccumulatorTest.java b/test/unit/org/apache/cassandra/utils/concurrent/AccumulatorTest.java index 9b34a232a5..da5bd32151 100644 --- a/test/unit/org/apache/cassandra/utils/concurrent/AccumulatorTest.java +++ b/test/unit/org/apache/cassandra/utils/concurrent/AccumulatorTest.java @@ -29,7 +29,7 @@ public class AccumulatorTest @Test public void testAddMoreThanCapacity() { - Accumulator accu = new Accumulator(4); + Accumulator accu = new Accumulator<>(4); accu.add(1); accu.add(2); @@ -50,7 +50,7 @@ public class AccumulatorTest @Test public void testIsEmptyAndSize() { - Accumulator accu = new Accumulator(4); + Accumulator accu = new Accumulator<>(4); assertTrue(accu.isEmpty()); assertEquals(0, accu.size()); @@ -58,20 +58,20 @@ public class AccumulatorTest accu.add(1); accu.add(2); - assertTrue(!accu.isEmpty()); + assertFalse(accu.isEmpty()); assertEquals(2, accu.size()); accu.add(3); accu.add(4); - assertTrue(!accu.isEmpty()); + assertFalse(accu.isEmpty()); assertEquals(4, accu.size()); } @Test public void testGetAndIterator() { - Accumulator accu = new Accumulator(4); + Accumulator accu = new Accumulator<>(4); accu.add("3"); accu.add("2"); @@ -99,32 +99,32 @@ public class AccumulatorTest @Test public void testClearUnsafe() { - Accumulator accu = new Accumulator<>(3); + Accumulator accu = new Accumulator<>(5); accu.add("1"); accu.add("2"); accu.add("3"); - accu.clearUnsafe(); + accu.clearUnsafe(1); - assertEquals(0, accu.size()); - assertFalse(accu.snapshot().iterator().hasNext()); - assertOutOfBonds(accu, 0); + assertEquals(3, accu.size()); + assertTrue(accu.snapshot().iterator().hasNext()); accu.add("4"); accu.add("5"); - assertEquals(2, accu.size()); + assertEquals(5, accu.size()); - assertEquals("4", accu.get(0)); - assertEquals("5", accu.get(1)); - assertOutOfBonds(accu, 2); + assertEquals("4", accu.get(3)); + assertEquals("5", accu.get(4)); + assertOutOfBonds(accu, 5); Iterator iter = accu.snapshot().iterator(); assertTrue(iter.hasNext()); - assertEquals("4", iter.next()); - assertEquals("5", iter.next()); - assertFalse(iter.hasNext()); + assertEquals("1", iter.next()); + assertNull(iter.next()); + assertTrue(iter.hasNext()); + assertEquals("3", iter.next()); } private static void assertOutOfBonds(Accumulator accumulator, int index)