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

Track top partitions by size and tombstone count 

Patch by marcuse; reviewed by David Capwell and Yifan Cai for CASSANDRA-16310
This commit is contained in:
Marcus Eriksson 2020-12-03 11:29:38 +01:00
parent 49cc352916
commit 545809616c
22 changed files with 1403 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
NEWS.txt
View File

@ -99,6 +99,10 @@ New features
any writes to the CDC-enabled tables will be blocked when reaching to the limit for CDC data on disk, which is the
existing and the default behavior. Setting to false, the writes to the CDC-enabled tables will be accepted and
the oldest CDC data on disk will be deleted to ensure the size constraint.
- Top partitions based on partition size or tombstone count are now tracked per table. These partitions are stored
in a new system.top_partitions table and exposed via JMX and nodetool tablestats. The partitions are tracked
during full or validation repairs but not incremental ones since those don't include all sstables and the partition
size/tombstone count would not be correct.
- New native functions to convert unix time values into C* native types: toDate(bigint), toTimestamp(bigint),
mintimeuuid(bigint) and maxtimeuuid(bigint)
- Support for multiple permission in a single GRANT/REVOKE/LIST statement has been added. It allows to

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

@ -1015,6 +1015,12 @@ public class Config
*/
public volatile int paxos_repair_parallelism = -1;
public volatile int max_top_size_partition_count = 10;
public volatile int max_top_tombstone_partition_count = 10;
public volatile DataStorageSpec min_tracked_partition_size_bytes = DataStorageSpec.inMebibytes(1);
public volatile long min_tracked_partition_tombstone_count = 5000;
public volatile boolean top_partitions_enabled = true;
public static Supplier<Config> getOverrideLoadConfig()
{
return overrideLoadConfig;

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

@ -4194,4 +4194,49 @@ public class DatabaseDescriptor
conf.repair_state_size = size;
}
}
public static boolean topPartitionsEnabled()
{
return conf.top_partitions_enabled;
}
public static int getMaxTopSizePartitionCount()
{
return conf.max_top_size_partition_count;
}
public static void setMaxTopSizePartitionCount(int value)
{
conf.max_top_size_partition_count = value;
}
public static int getMaxTopTombstonePartitionCount()
{
return conf.max_top_tombstone_partition_count;
}
public static void setMaxTopTombstonePartitionCount(int value)
{
conf.max_top_tombstone_partition_count = value;
}
public static DataStorageSpec getMinTrackedPartitionSize()
{
return conf.min_tracked_partition_size_bytes;
}
public static void setMinTrackedPartitionSize(DataStorageSpec spec)
{
conf.min_tracked_partition_size_bytes = spec;
}
public static long getMinTrackedPartitionTombstoneCount()
{
return conf.min_tracked_partition_tombstone_count;
}
public static void setMinTrackedPartitionTombstoneCount(long value)
{
conf.min_tracked_partition_tombstone_count = value;
}
}

43
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -126,6 +126,7 @@ import org.apache.cassandra.metrics.Sampler;
import org.apache.cassandra.metrics.Sampler.Sample;
import org.apache.cassandra.metrics.Sampler.SamplerType;
import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.repair.TableRepairManager;
import org.apache.cassandra.repair.consistent.admin.CleanupSummary;
import org.apache.cassandra.repair.consistent.admin.PendingStat;
@ -149,7 +150,7 @@ import org.apache.cassandra.service.paxos.TablePaxosRepairHistory;
import org.apache.cassandra.service.snapshot.SnapshotManifest;
import org.apache.cassandra.service.snapshot.TableSnapshot;
import org.apache.cassandra.streaming.TableStreamManager;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.DefaultValue;
import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
@ -278,6 +279,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
private final CassandraStreamManager streamManager;
private final TableRepairManager repairManager;
public final TopPartitionTracker topPartitions;
private final SSTableImporter sstableImporter;
@ -543,6 +545,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
streamManager = new CassandraStreamManager(this);
repairManager = new CassandraTableRepairManager(this);
sstableImporter = new SSTableImporter(this);
if (SchemaConstants.isSystemKeyspace(keyspace.getName()))
topPartitions = null;
else
topPartitions = new TopPartitionTracker(metadata());
}
public static String getTableMBeanName(String ks, String name, boolean isIndex)
@ -665,6 +672,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
indexManager.dropAllIndexes(dropData);
invalidateCaches();
if (topPartitions != null)
topPartitions.close();
}
/**
@ -3155,4 +3164,36 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
return false;
}
@Override
public Map<String, Long> getTopSizePartitions()
{
if (topPartitions == null)
return Collections.emptyMap();
return topPartitions.getTopSizePartitionMap();
}
@Override
public Long getTopSizePartitionsLastUpdate()
{
if (topPartitions == null)
return null;
return topPartitions.topSizes().lastUpdate;
}
@Override
public Map<String, Long> getTopTombstonePartitions()
{
if (topPartitions == null)
return Collections.emptyMap();
return topPartitions.getTopTombstonePartitionMap();
}
@Override
public Long getTopTombstonePartitionsLastUpdate()
{
if (topPartitions == null)
return null;
return topPartitions.topTombstones().lastUpdate;
}
}

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

@ -266,4 +266,9 @@ public interface ColumnFamilyStoreMBean
public boolean hasMisplacedSSTables();
public List<String> getDataPaths() throws IOException;
public Map<String, Long> getTopSizePartitions();
public Long getTopSizePartitionsLastUpdate();
public Map<String, Long> getTopTombstonePartitions();
public Long getTopTombstonePartitionsLastUpdate();
}

66
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -27,6 +27,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@ -45,6 +46,7 @@ import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
import com.google.common.io.ByteStreams;
@ -64,7 +66,9 @@ import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.compaction.CompactionHistoryTabularData;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.db.marshal.TupleType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.db.partitions.PartitionUpdate;
@ -84,6 +88,7 @@ import org.apache.cassandra.io.util.RebufferingInputStream;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.RestorableMeter;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.CompactionParams;
import org.apache.cassandra.schema.Functions;
@ -159,6 +164,7 @@ public final class SystemKeyspace
public static final String BUILT_VIEWS = "built_views";
public static final String PREPARED_STATEMENTS = "prepared_statements";
public static final String REPAIRS = "repairs";
public static final String TOP_PARTITIONS = "top_partitions";
/**
* By default the system keyspace tables should be stored in a single data directory to allow the server
@ -400,6 +406,19 @@ public final class SystemKeyspace
+ "PRIMARY KEY ((keyspace_name), view_name))")
.build();
private static final TableMetadata TopPartitions =
parse(TOP_PARTITIONS,
"Stores the top partitions",
"CREATE TABLE %s ("
+ "keyspace_name text,"
+ "table_name text,"
+ "top_type text,"
+ "top frozen<list<tuple<text, bigint>>>,"
+ "last_update timestamp,"
+ "PRIMARY KEY (keyspace_name, table_name, top_type))")
.build();
private static final TableMetadata PreparedStatements =
parse(PREPARED_STATEMENTS,
"prepared statements",
@ -513,7 +532,8 @@ public final class SystemKeyspace
ViewBuildsInProgress,
BuiltViews,
PreparedStatements,
Repairs);
Repairs,
TopPartitions);
}
private static Functions functions()
@ -1823,4 +1843,48 @@ public final class SystemKeyspace
public static interface TriFunction<A, B, C, D> {
D accept(A var1, B var2, C var3);
}
public static void saveTopPartitions(TableMetadata metadata, String topType, Collection<TopPartitionTracker.TopPartition> topPartitions, long lastUpdate)
{
String cql = String.format("INSERT INTO %s.%s (keyspace_name, table_name, top_type, top, last_update) values (?, ?, ?, ?, ?)", SchemaConstants.SYSTEM_KEYSPACE_NAME, TOP_PARTITIONS);
List<ByteBuffer> tupleList = new ArrayList<>(topPartitions.size());
topPartitions.forEach(tp -> {
String key = metadata.partitionKeyType.getString(tp.key.getKey());
tupleList.add(TupleType.buildValue(new ByteBuffer[] { UTF8Type.instance.decompose(key),
LongType.instance.decompose(tp.value)}));
});
executeInternal(cql, metadata.keyspace, metadata.name, topType, tupleList, Date.from(Instant.ofEpochMilli(lastUpdate)));
}
public static TopPartitionTracker.StoredTopPartitions getTopPartitions(TableMetadata metadata, String topType)
{
try
{
String cql = String.format("SELECT top, last_update FROM %s.%s WHERE keyspace_name = ? and table_name = ? and top_type = ?", SchemaConstants.SYSTEM_KEYSPACE_NAME, TOP_PARTITIONS);
UntypedResultSet res = executeInternal(cql, metadata.keyspace, metadata.name, topType);
if (res == null || res.isEmpty())
return TopPartitionTracker.StoredTopPartitions.EMPTY;
UntypedResultSet.Row row = res.one();
long lastUpdated = row.getLong("last_update");
List<ByteBuffer> top = row.getList("top", BytesType.instance);
if (top == null || top.isEmpty())
return TopPartitionTracker.StoredTopPartitions.EMPTY;
List<TopPartitionTracker.TopPartition> topPartitions = new ArrayList<>(top.size());
TupleType tupleType = new TupleType(Lists.newArrayList(UTF8Type.instance, LongType.instance));
for (ByteBuffer bb : top)
{
ByteBuffer[] components = tupleType.split(bb);
String keyStr = UTF8Type.instance.compose(components[0]);
long value = LongType.instance.compose(components[1]);
topPartitions.add(new TopPartitionTracker.TopPartition(metadata.partitioner.decorateKey(metadata.partitionKeyType.fromString(keyStr)), value));
}
return new TopPartitionTracker.StoredTopPartitions(topPartitions, lastUpdated);
}
catch (Exception e)
{
logger.warn("Could not load stored top {} partitions for {}.{}", topType, metadata.keyspace, metadata.name, e);
return TopPartitionTracker.StoredTopPartitions.EMPTY;
}
}
}

13
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -42,6 +42,7 @@ import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.index.transactions.CompactionTransaction;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
import org.apache.cassandra.service.paxos.PaxosRepairHistory;
import org.apache.cassandra.service.paxos.uncommitted.PaxosRows;
@ -94,11 +95,17 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
public CompactionIterator(OperationType type, List<ISSTableScanner> scanners, AbstractCompactionController controller, int nowInSec, TimeUUID compactionId)
{
this(type, scanners, controller, nowInSec, compactionId, ActiveCompactionsTracker.NOOP);
this(type, scanners, controller, nowInSec, compactionId, ActiveCompactionsTracker.NOOP, null);
}
@SuppressWarnings("resource") // We make sure to close mergedIterator in close() and CompactionIterator is itself an AutoCloseable
public CompactionIterator(OperationType type, List<ISSTableScanner> scanners, AbstractCompactionController controller, int nowInSec, TimeUUID compactionId, ActiveCompactionsTracker activeCompactions)
public CompactionIterator(OperationType type,
List<ISSTableScanner> scanners,
AbstractCompactionController controller,
int nowInSec,
TimeUUID compactionId,
ActiveCompactionsTracker activeCompactions,
TopPartitionTracker.Collector topPartitionCollector)
{
this.controller = controller;
this.type = type;
@ -121,6 +128,8 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
UnfilteredPartitionIterator merged = scanners.isEmpty()
? EmptyIterators.unfilteredPartition(controller.cfs.metadata())
: UnfilteredPartitionIterators.merge(scanners, listener());
if (topPartitionCollector != null) // need to count tombstones before they are purged
merged = Transformation.apply(merged, new TopPartitionTracker.TombstoneCounter(topPartitionCollector, nowInSec));
merged = Transformation.apply(merged, new GarbageSkipper(controller));
Transformation<UnfilteredRowIterator> purger = isPaxos(controller.cfs) && paxosStatePurging() != legacy
? new PaxosPurger(nowInSec)

6
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -1310,7 +1310,7 @@ public class CompactionManager implements CompactionManagerMBean
ISSTableScanner scanner = cleanupStrategy.getScanner(sstable);
CompactionController controller = new CompactionController(cfs, txn.originals(), getDefaultGcBefore(cfs, nowInSec));
Refs<SSTableReader> refs = Refs.ref(Collections.singleton(sstable));
CompactionIterator ci = new CompactionIterator(OperationType.CLEANUP, Collections.singletonList(scanner), controller, nowInSec, nextTimeUUID(), active))
CompactionIterator ci = new CompactionIterator(OperationType.CLEANUP, Collections.singletonList(scanner), controller, nowInSec, nextTimeUUID(), active, null))
{
StatsMetadata metadata = sstable.getSSTableMetadata();
writer.switchWriter(createWriter(cfs, compactionFileLocation, expectedBloomFilterSize, metadata.repairedAt, metadata.pendingRepair, metadata.isTransient, sstable, txn));
@ -1732,8 +1732,8 @@ public class CompactionManager implements CompactionManagerMBean
@VisibleForTesting
public static CompactionIterator getAntiCompactionIterator(List<ISSTableScanner> scanners, CompactionController controller, int nowInSec, TimeUUID timeUUID, ActiveCompactionsTracker activeCompactions, BooleanSupplier isCancelled)
{
return new CompactionIterator(OperationType.ANTICOMPACTION, scanners, controller, nowInSec, timeUUID, activeCompactions) {
return new CompactionIterator(OperationType.ANTICOMPACTION, scanners, controller, nowInSec, timeUUID, activeCompactions, null)
{
public boolean isStopRequested()
{
return super.isStopRequested() || isCancelled.getAsBoolean();

5
src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java
View File

@ -31,6 +31,7 @@ import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.repair.TableRepairManager;
import org.apache.cassandra.repair.ValidationPartitionIterator;
import org.apache.cassandra.repair.NoSuchRepairSessionException;
@ -46,9 +47,9 @@ public class CassandraTableRepairManager implements TableRepairManager
}
@Override
public ValidationPartitionIterator getValidationIterator(Collection<Range<Token>> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec) throws IOException, NoSuchRepairSessionException
public ValidationPartitionIterator getValidationIterator(Collection<Range<Token>> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec, TopPartitionTracker.Collector topPartitionCollector) throws IOException, NoSuchRepairSessionException
{
return new CassandraValidationIterator(cfs, ranges, parentId, sessionID, isIncremental, nowInSec);
return new CassandraValidationIterator(cfs, ranges, parentId, sessionID, isIncremental, nowInSec, topPartitionCollector);
}
@Override

9
src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
View File

@ -51,6 +51,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.repair.ValidationPartitionIterator;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ActiveRepairService;
@ -105,9 +106,9 @@ public class CassandraValidationIterator extends ValidationPartitionIterator
private static class ValidationCompactionIterator extends CompactionIterator
{
public ValidationCompactionIterator(List<ISSTableScanner> scanners, ValidationCompactionController controller, int nowInSec, ActiveCompactionsTracker activeCompactions)
public ValidationCompactionIterator(List<ISSTableScanner> scanners, ValidationCompactionController controller, int nowInSec, ActiveCompactionsTracker activeCompactions, TopPartitionTracker.Collector topPartitionCollector)
{
super(OperationType.VALIDATION, scanners, controller, nowInSec, nextTimeUUID(), activeCompactions);
super(OperationType.VALIDATION, scanners, controller, nowInSec, nextTimeUUID(), activeCompactions, topPartitionCollector);
}
}
@ -172,7 +173,7 @@ public class CassandraValidationIterator extends ValidationPartitionIterator
private final long estimatedPartitions;
private final Map<Range<Token>, Long> rangePartitionCounts;
public CassandraValidationIterator(ColumnFamilyStore cfs, Collection<Range<Token>> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec) throws IOException, NoSuchRepairSessionException
public CassandraValidationIterator(ColumnFamilyStore cfs, Collection<Range<Token>> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec, TopPartitionTracker.Collector topPartitionCollector) throws IOException, NoSuchRepairSessionException
{
this.cfs = cfs;
@ -211,7 +212,7 @@ public class CassandraValidationIterator extends ValidationPartitionIterator
controller = new ValidationCompactionController(cfs, getDefaultGcBefore(cfs, nowInSec));
scanners = cfs.getCompactionStrategyManager().getScanners(sstables, ranges);
ci = new ValidationCompactionIterator(scanners.scanners, controller, nowInSec, CompactionManager.instance.active);
ci = new ValidationCompactionIterator(scanners.scanners, controller, nowInSec, CompactionManager.instance.active, topPartitionCollector);
long allPartitions = 0;
rangePartitionCounts = Maps.newHashMapWithExpectedSize(ranges.size());

408
src/java/org/apache/cassandra/metrics/TopPartitionTracker.java Normal file
View File

@ -0,0 +1,408 @@
/*
* 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.metrics;
import java.io.Closeable;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.TreeSet;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.RangeTombstoneMarker;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.Clock;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Tracks top partitions, currently by size and by tombstone count
*
* Collects during full and preview (-vd) repair since then we read the full partition
*
* Note that since we can run sub range repair there might be windows where the top partitions are not correct -
* for example, assume we track the top 2 partitions for this node:
*
* tokens with size:
* (a, 100); (b, 40); (c, 10); (d, 100); (e, 50); (f, 10)
* - top2: a, d
* now a is deleted and we run a repair for keys [a, c]
* - top2: b, d
* and when we repair [d, f]
* - top2: d, e
*
*/
public class TopPartitionTracker implements Closeable
{
private final static String SIZES = "SIZES";
private final static String TOMBSTONES = "TOMBSTONES";
private final AtomicReference<TopHolder> topSizes = new AtomicReference<>();
private final AtomicReference<TopHolder> topTombstones = new AtomicReference<>();
private final TableMetadata metadata;
private final Future<?> scheduledSave;
private long lastTombstoneSave = 0;
private long lastSizeSave = 0;
public TopPartitionTracker(TableMetadata metadata)
{
this.metadata = metadata;
topSizes.set(new TopHolder(SystemKeyspace.getTopPartitions(metadata, SIZES),
DatabaseDescriptor.getMaxTopSizePartitionCount(),
DatabaseDescriptor.getMinTrackedPartitionSize().toBytes()));
topTombstones.set(new TopHolder(SystemKeyspace.getTopPartitions(metadata, TOMBSTONES),
DatabaseDescriptor.getMaxTopTombstonePartitionCount(),
DatabaseDescriptor.getMinTrackedPartitionTombstoneCount()));
scheduledSave = ScheduledExecutors.optionalTasks.scheduleAtFixedRate(this::save, 60, 60, TimeUnit.MINUTES);
}
public void close()
{
scheduledSave.cancel(true);
}
@VisibleForTesting
public void save()
{
TopHolder sizes = topSizes.get();
if (!sizes.top.isEmpty() && sizes.lastUpdate > lastSizeSave)
{
SystemKeyspace.saveTopPartitions(metadata, SIZES, sizes.top, sizes.lastUpdate);
lastSizeSave = sizes.lastUpdate;
}
TopHolder tombstones = topTombstones.get();
if (!tombstones.top.isEmpty() && tombstones.lastUpdate > lastTombstoneSave)
{
SystemKeyspace.saveTopPartitions(metadata, TOMBSTONES, tombstones.top, tombstones.lastUpdate);
lastTombstoneSave = tombstones.lastUpdate;
}
}
public void merge(Collector collector)
{
while (true)
{
TopHolder cur = topSizes.get();
TopHolder newSizes = cur.merge(collector.sizes, StorageService.instance.getLocalReplicas(metadata.keyspace).ranges());
if (topSizes.compareAndSet(cur, newSizes))
break;
}
while (true)
{
TopHolder cur = topTombstones.get();
TopHolder newTombstones = cur.merge(collector.tombstones, StorageService.instance.getLocalReplicas(metadata.keyspace).ranges());
if (topTombstones.compareAndSet(cur, newTombstones))
break;
}
}
@Override
public String toString()
{
return "TopPartitionTracker:\n" +
"topSizes:\n" + topSizes.get() + '\n' +
"topTombstones:\n" + topTombstones.get() + '\n';
}
public Map<String, Long> getTopTombstonePartitionMap()
{
return topTombstones.get().toMap(metadata);
}
public Map<String, Long> getTopSizePartitionMap()
{
return topSizes.get().toMap(metadata);
}
@VisibleForTesting
public TopHolder topSizes()
{
return topSizes.get();
}
@VisibleForTesting
public TopHolder topTombstones()
{
return topTombstones.get();
}
public static class Collector
{
private final TopHolder tombstones;
private final TopHolder sizes;
public Collector(Collection<Range<Token>> ranges)
{
this.tombstones = new TopHolder(DatabaseDescriptor.getMaxTopTombstonePartitionCount(),
DatabaseDescriptor.getMinTrackedPartitionTombstoneCount(),
ranges);
this.sizes = new TopHolder(DatabaseDescriptor.getMaxTopSizePartitionCount(),
DatabaseDescriptor.getMinTrackedPartitionSize().toBytes(),
ranges);
}
public void trackTombstoneCount(DecoratedKey key, long count)
{
tombstones.track(key, count);
}
public void trackPartitionSize(DecoratedKey key, long size)
{
sizes.track(key, size);
}
public String toString()
{
return "tombstones:\n"+tombstones+"\nsizes:\n"+sizes;
}
}
public static class TopHolder
{
public final NavigableSet<TopPartition> top;
private final int maxTopPartitionCount;
private final long minTrackedValue;
private final Collection<Range<Token>> ranges;
private long currentMinValue = Long.MAX_VALUE;
public final long lastUpdate;
private TopHolder(int maxTopPartitionCount, long minTrackedValue, Collection<Range<Token>> ranges)
{
this(maxTopPartitionCount, minTrackedValue, new TreeSet<>(), ranges, 0);
}
private TopHolder(int maxTopPartitionCount, long minTrackedValue, NavigableSet<TopPartition> top, Collection<Range<Token>> ranges, long lastUpdate)
{
this.maxTopPartitionCount = maxTopPartitionCount;
this.minTrackedValue = minTrackedValue;
this.top = top;
this.ranges = ranges;
this.lastUpdate = lastUpdate;
}
private TopHolder(StoredTopPartitions storedTopPartitions,
int maxTopPartitionCount,
long minTrackedValue)
{
this.maxTopPartitionCount = maxTopPartitionCount;
this.minTrackedValue = minTrackedValue;
top = new TreeSet<>();
this.ranges = null;
this.lastUpdate = storedTopPartitions.lastUpdated;
for (TopPartition topPartition : storedTopPartitions.topPartitions)
track(topPartition);
}
public void track(DecoratedKey key, long value)
{
if (value < minTrackedValue)
return;
if (top.size() < maxTopPartitionCount || value > currentMinValue)
track(new TopPartition(SSTable.getMinimalKey(key), value));
}
private void track(TopPartition tp)
{
top.add(tp);
while (top.size() > maxTopPartitionCount)
{
top.pollLast();
currentMinValue = top.last().value;
}
currentMinValue = Math.min(tp.value, currentMinValue);
}
/**
* we merge any pre-existing top partitions on to the ones we just collected if they are outside of the
* range collected.
*
* This means that if a large partition is deleted it will disappear from the top partitions
*
* @param holder the newly collected holder - this will get copied and any existing token outside of the collected ranges will get added to the copy
* @param ownedRanges the ranges this node owns - any existing token outside of these ranges will get dropped
*/
public TopHolder merge(TopHolder holder, Collection<Range<Token>> ownedRanges)
{
TopHolder mergedHolder = holder.cloneForMerging(currentTimeMillis());
for (TopPartition existingTop : top)
{
if (!Range.isInRanges(existingTop.key.getToken(), mergedHolder.ranges) &&
(ownedRanges.isEmpty() || Range.isInRanges(existingTop.key.getToken(), ownedRanges))) // make sure we drop any tokens that we don't own anymore
mergedHolder.track(existingTop);
}
return mergedHolder;
}
private TopHolder cloneForMerging(long lastUpdate)
{
return new TopHolder(maxTopPartitionCount, minTrackedValue, new TreeSet<>(top), ranges, lastUpdate);
}
public String toString()
{
int i = 0;
Iterator<TopPartition> it = top.iterator();
StringBuilder sb = new StringBuilder();
while (it.hasNext())
{
i++;
sb.append(i).append(':').append(it.next()).append(System.lineSeparator());
}
return sb.toString();
}
public Map<String, Long> toMap(TableMetadata metadata)
{
Map<String, Long> topPartitionsMap = new LinkedHashMap<>();
for (TopPartitionTracker.TopPartition topPartition : top)
{
String key = metadata.partitionKeyType.getString(topPartition.key.getKey());
topPartitionsMap.put(key, topPartition.value);
}
return topPartitionsMap;
}
}
private static final Comparator<TopPartition> comparator = (o1, o2) -> {
int cmp = -Long.compare(o1.value, o2.value);
if (cmp != 0) return cmp;
return o1.key.compareTo(o2.key);
};
public static class TopPartition implements Comparable<TopPartition>
{
public final DecoratedKey key;
public final long value;
public TopPartition(DecoratedKey key, long value)
{
this.key = key;
this.value = value;
}
@Override
public int compareTo(TopPartition o)
{
return comparator.compare(this, o);
}
@Override
public String toString()
{
return "TopPartition{" +
"key=" + key +
", value=" + value +
'}';
}
}
public static class TombstoneCounter extends Transformation<UnfilteredRowIterator>
{
private final TopPartitionTracker.Collector collector;
private final int nowInSec;
private long tombstoneCount = 0;
private DecoratedKey key = null;
public TombstoneCounter(TopPartitionTracker.Collector collector, int nowInSec)
{
this.collector = collector;
this.nowInSec = nowInSec;
}
@Override
public Row applyToRow(Row row)
{
if (!row.deletion().isLive())
tombstoneCount++;
if (row.hasDeletion(nowInSec))
{
for (Cell<?> c : row.cells())
if (c.isTombstone())
tombstoneCount++;
}
return row;
}
@Override
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
tombstoneCount++;
return marker;
}
@Override
protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
reset(partition.partitionKey());
if (!partition.partitionLevelDeletion().isLive())
tombstoneCount++;
return Transformation.apply(partition, this);
}
private void reset(DecoratedKey key)
{
tombstoneCount = 0;
this.key = key;
}
@Override
public void onPartitionClose()
{
collector.trackTombstoneCount(key, tombstoneCount);
}
}
public static class StoredTopPartitions
{
public static StoredTopPartitions EMPTY = new StoredTopPartitions(Collections.emptyList(), 0);
public final List<TopPartition> topPartitions;
public final long lastUpdated;
public StoredTopPartitions(List<TopPartition> topPartitions, long lastUpdated)
{
this.topPartitions = topPartitions;
this.lastUpdated = lastUpdated;
}
}
}

3
src/java/org/apache/cassandra/repair/TableRepairManager.java
View File

@ -25,6 +25,7 @@ import java.util.concurrent.Future;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.utils.TimeUUID;
/**
@ -37,7 +38,7 @@ public interface TableRepairManager
* data previously isolated for repair with the given parentId. nowInSec should determine whether tombstones should
* be purged or not.
*/
ValidationPartitionIterator getValidationIterator(Collection<Range<Token>> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec) throws IOException, NoSuchRepairSessionException;
ValidationPartitionIterator getValidationIterator(Collection<Range<Token>> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec, TopPartitionTracker.Collector topPartitionCollector) throws IOException, NoSuchRepairSessionException;
/**
* Begin execution of the given validation callable. Which thread pool a validation should run in is an implementation detail.

32
src/java/org/apache/cassandra/repair/ValidationManager.java
View File

@ -35,6 +35,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.repair.state.ValidationState;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
@ -87,10 +88,10 @@ public class ValidationManager
return trees;
}
private static ValidationPartitionIterator getValidationIterator(TableRepairManager repairManager, Validator validator) throws IOException, NoSuchRepairSessionException
private static ValidationPartitionIterator getValidationIterator(TableRepairManager repairManager, Validator validator, TopPartitionTracker.Collector topPartitionCollector) throws IOException, NoSuchRepairSessionException
{
RepairJobDesc desc = validator.desc;
return repairManager.getValidationIterator(desc.ranges, desc.parentSessionId, desc.sessionId, validator.isIncremental, validator.nowInSec);
return repairManager.getValidationIterator(desc.ranges, desc.parentSessionId, desc.sessionId, validator.isIncremental, validator.nowInSec, topPartitionCollector);
}
/**
@ -112,15 +113,19 @@ public class ValidationManager
return;
}
TopPartitionTracker.Collector topPartitionCollector = null;
if (cfs.topPartitions != null && DatabaseDescriptor.topPartitionsEnabled() && isTopPartitionSupported(validator))
topPartitionCollector = new TopPartitionTracker.Collector(validator.desc.ranges);
// Create Merkle trees suitable to hold estimated partitions for the given ranges.
// We blindly assume that a partition is evenly distributed on all sstables for now.
long start = nanoTime();
try (ValidationPartitionIterator vi = getValidationIterator(cfs.getRepairManager(), validator))
try (ValidationPartitionIterator vi = getValidationIterator(cfs.getRepairManager(), validator, topPartitionCollector))
{
state.phase.start(vi.estimatedPartitions(), vi.getEstimatedBytes());
MerkleTrees trees = createMerkleTrees(vi, validator.desc.ranges, cfs);
// validate the CF as we iterate over it
validator.prepare(cfs, trees);
validator.prepare(cfs, trees, topPartitionCollector);
while (vi.hasNext())
{
try (UnfilteredRowIterator partition = vi.next())
@ -138,6 +143,8 @@ public class ValidationManager
{
cfs.metric.bytesValidated.update(state.estimatedTotalBytes);
cfs.metric.partitionsValidated.update(state.partitionsProcessed);
if (topPartitionCollector != null)
cfs.topPartitions.merge(topPartitionCollector);
}
if (logger.isDebugEnabled())
{
@ -150,6 +157,23 @@ public class ValidationManager
}
}
private static boolean isTopPartitionSupported(Validator validator)
{
// supported: --validate, --full, --full --preview
switch (validator.getPreviewKind())
{
case NONE:
return !validator.isIncremental;
case ALL:
case REPAIRED:
return true;
case UNREPAIRED:
return false;
default:
throw new AssertionError("Unknown preview kind: " + validator.getPreviewKind());
}
}
/**
* Does not mutate data, so is not scheduled.
*/

12
src/java/org/apache/cassandra/repair/Validator.java
View File

@ -34,6 +34,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.ValidationResponse;
@ -78,6 +79,7 @@ public class Validator implements Runnable
private final PreviewKind previewKind;
public final ValidationState state;
public TopPartitionTracker.Collector topPartitionCollector;
public Validator(ValidationState state, int nowInSec, PreviewKind previewKind)
{
@ -103,9 +105,10 @@ public class Validator implements Runnable
this.evenTreeDistribution = evenTreeDistribution;
}
public void prepare(ColumnFamilyStore cfs, MerkleTrees trees)
public void prepare(ColumnFamilyStore cfs, MerkleTrees trees, TopPartitionTracker.Collector topPartitionCollector)
{
this.trees = trees;
this.topPartitionCollector = topPartitionCollector;
if (!trees.partitioner().preservesOrder() || evenTreeDistribution)
{
@ -177,6 +180,8 @@ public class Validator implements Runnable
RowHash rowHash = rowHash(partition);
if (rowHash != null)
{
if(topPartitionCollector != null)
topPartitionCollector.trackPartitionSize(partition.partitionKey(), rowHash.size);
range.addHash(rowHash);
}
}
@ -254,6 +259,11 @@ public class Validator implements Runnable
respond(new ValidationResponse(desc, trees));
}
public PreviewKind getPreviewKind()
{
return previewKind;
}
private boolean initiatorIsRemote()
{
return !FBUtilities.getBroadcastAddressAndPort().equals(initiator);

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

@ -6710,4 +6710,58 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
{
return PaxosRepair.getSkipPaxosRepairCompatibilityCheck();
}
@Override
public boolean topPartitionsEnabled()
{
return DatabaseDescriptor.topPartitionsEnabled();
}
@Override
public int getMaxTopSizePartitionCount()
{
return DatabaseDescriptor.getMaxTopSizePartitionCount();
}
@Override
public void setMaxTopSizePartitionCount(int value)
{
DatabaseDescriptor.setMaxTopSizePartitionCount(value);
}
@Override
public int getMaxTopTombstonePartitionCount()
{
return DatabaseDescriptor.getMaxTopTombstonePartitionCount();
}
@Override
public void setMaxTopTombstonePartitionCount(int value)
{
DatabaseDescriptor.setMaxTopTombstonePartitionCount(value);
}
@Override
public String getMinTrackedPartitionSize()
{
return DatabaseDescriptor.getMinTrackedPartitionSize().toString();
}
@Override
public void setMinTrackedPartitionSize(String value)
{
DatabaseDescriptor.setMinTrackedPartitionSize(parseDataStorageSpec(value));
}
@Override
public long getMinTrackedPartitionTombstoneCount()
{
return DatabaseDescriptor.getMinTrackedPartitionTombstoneCount();
}
@Override
public void setMinTrackedPartitionTombstoneCount(long value)
{
DatabaseDescriptor.setMinTrackedPartitionTombstoneCount(value);
}
}

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

@ -991,4 +991,13 @@ public interface StorageServiceMBean extends NotificationEmitter
public boolean getSkipPaxosRepairCompatibilityCheck();
String getToken(String keyspaceName, String table, String partitionKey);
public boolean topPartitionsEnabled();
public int getMaxTopSizePartitionCount();
public void setMaxTopSizePartitionCount(int value);
public int getMaxTopTombstonePartitionCount();
public void setMaxTopTombstonePartitionCount(int value);
public String getMinTrackedPartitionSize();
public void setMinTrackedPartitionSize(String value);
public long getMinTrackedPartitionTombstoneCount();
public void setMinTrackedPartitionTombstoneCount(long value);
}

5
src/java/org/apache/cassandra/tools/nodetool/stats/StatsTable.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.tools.nodetool.stats;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
public class StatsTable
{
@ -72,4 +73,8 @@ public class StatsTable
public List<String> sstableBytesInEachLevel = new ArrayList<>();
public Boolean isInCorrectLocation = null; // null: option not active
public double droppableTombstoneRatio;
public Map<String, String> topSizePartitions;
public Map<String, Long> topTombstonePartitions;
public String topSizePartitionsLastUpdate;
public String topTombstonePartitionsLastUpdate;
}

27
src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsHolder.java
View File

@ -18,6 +18,8 @@
package org.apache.cassandra.tools.nodetool.stats;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import com.google.common.collect.ArrayListMultimap;
@ -164,6 +166,8 @@ public class TableStatsHolder implements StatsHolder
mpTable.put("dropped_mutations", table.droppedMutations);
mpTable.put("droppable_tombstone_ratio",
String.format("%01.5f", table.droppableTombstoneRatio));
mpTable.put("top_size_partitions", table.topSizePartitions);
mpTable.put("top_tombstone_partitions", table.topTombstonePartitions);
if (locationCheck)
mpTable.put("sstables_in_correct_location", table.isInCorrectLocation);
return mpTable;
@ -360,6 +364,13 @@ public class TableStatsHolder implements StatsHolder
statsTable.maximumTombstonesPerSliceLastFiveMinutes = histogram.getMax();
statsTable.droppedMutations = format((Long) probe.getColumnFamilyMetric(keyspaceName, tableName, "DroppedMutations"), humanReadable);
statsTable.droppableTombstoneRatio = probe.getDroppableTombstoneRatio(keyspaceName, tableName);
statsTable.topSizePartitions = format(table.getTopSizePartitions(), humanReadable);
if (table.getTopSizePartitionsLastUpdate() != null)
statsTable.topSizePartitionsLastUpdate = millisToDateString(table.getTopSizePartitionsLastUpdate());
statsTable.topTombstonePartitions = table.getTopTombstonePartitions();
if (table.getTopTombstonePartitionsLastUpdate() != null)
statsTable.topTombstonePartitionsLastUpdate = millisToDateString(table.getTopTombstonePartitionsLastUpdate());
statsKeyspace.tables.add(statsTable);
}
keyspaces.add(statsKeyspace);
@ -371,6 +382,22 @@ public class TableStatsHolder implements StatsHolder
return humanReadable ? FileUtils.stringifyFileSize(bytes) : Long.toString(bytes);
}
private Map<String, String> format(Map<String, Long> map, boolean humanReadable)
{
LinkedHashMap<String, String> retMap = new LinkedHashMap<>();
for (Map.Entry<String, Long> entry : map.entrySet())
retMap.put(entry.getKey(), format(entry.getValue(), humanReadable));
return retMap;
}
private String millisToDateString(long millis)
{
TimeZone tz = TimeZone.getTimeZone("UTC");
DateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
df.setTimeZone(tz);
return df.format(new Date(millis));
}
/**
* Sort and filter this TableStatHolder's tables as specified by its sortKey and top attributes.
*/

18
src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsPrinter.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.tools.nodetool.stats;
import java.io.PrintStream;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.utils.FBUtilities;
@ -133,6 +134,23 @@ public class TableStatsPrinter<T extends StatsHolder>
out.printf(indent + "Droppable tombstone ratio: %01.5f%n", table.droppableTombstoneRatio);
if (table.isInCorrectLocation != null)
out.println(indent + "SSTables in correct location: " + table.isInCorrectLocation);
if (table.topSizePartitions != null && !table.topSizePartitions.isEmpty())
{
out.printf(indent + "Top partitions by size (last update: %s):%n", table.topSizePartitionsLastUpdate);
int maxWidth = Math.max(table.topSizePartitions.keySet().stream().map(String::length).max(Integer::compareTo).get() + 3, 5);
out.printf(indent + " %-" + maxWidth + "s %s%n", "Key", "Size");
for (Map.Entry<String, String> size : table.topSizePartitions.entrySet())
out.printf(indent + " %-" + maxWidth + "s %s%n", size.getKey(), size.getValue());
}
if (table.topTombstonePartitions != null && !table.topTombstonePartitions.isEmpty())
{
out.printf(indent + "Top partitions by tombstone count (last update: %s):%n", table.topTombstonePartitionsLastUpdate);
int maxWidth = Math.max(table.topTombstonePartitions.keySet().stream().map(String::length).max(Integer::compareTo).get() + 3, 5);
out.printf(indent + " %-" + maxWidth + "s %s%n", "Key", "Count");
for (Map.Entry<String, Long> tombstonecnt : table.topTombstonePartitions.entrySet())
out.printf(indent + " %-" + maxWidth + "s %s%n", tombstonecnt.getKey(), tombstonecnt.getValue());
}
out.println("");
}
}

317
test/distributed/org/apache/cassandra/distributed/test/TopPartitionsTest.java Normal file
View File

@ -0,0 +1,317 @@
/*
* 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.Collection;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.assertj.core.api.Assertions;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.junit.Assert.assertEquals;
import static org.psjava.util.AssertStatus.assertTrue;
@RunWith(Parameterized.class)
public class TopPartitionsTest extends TestBaseImpl
{
public enum Repair
{
Incremental, Full, FullPreview
}
private static AtomicInteger COUNTER = new AtomicInteger(0);
private static Cluster CLUSTER;
private final Repair repair;
public TopPartitionsTest(Repair repair)
{
this.repair = repair;
}
@Parameterized.Parameters(name = "{0}")
public static Collection<Object[]> messages()
{
return Stream.of(Repair.values())
.map(a -> new Object[]{ a })
.collect(Collectors.toList());
}
@BeforeClass
public static void setup() throws IOException
{
CLUSTER = init(Cluster.build(2).withConfig(config ->
config.set("min_tracked_partition_size_bytes", "0MiB")
.set("min_tracked_partition_tombstone_count", 0)
.with(GOSSIP, NETWORK))
.start());
}
@AfterClass
public static void cleanup()
{
if (CLUSTER != null)
CLUSTER.close();
}
@Before
public void before()
{
setCount(10, 10);
}
@Test
public void basicPartitionSizeTest()
{
String name = "tbl" + COUNTER.getAndIncrement();
String table = KEYSPACE + "." + name;
CLUSTER.schemaChange("create table " + table + " (id int, ck int, t int, primary key (id, ck))");
for (int i = 0; i < 100; i++)
for (int j = 0; j < i; j++)
CLUSTER.coordinator(1).execute("insert into " + table + " (id, ck, t) values (?,?,?)", ConsistencyLevel.ALL, i, j, i * j + 100);
repair();
CLUSTER.forEach(inst -> inst.runOnInstance(() -> {
// partitions 99 -> 90 are the largest, make sure they are in the map;
Map<String, Long> sizes = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopSizePartitions();
for (int i = 99; i >= 90; i--)
assertTrue(sizes.containsKey(String.valueOf(i)));
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
assertEquals(10, tombstones.size());
assertTrue(tombstones.values().stream().allMatch(l -> l == 0));
}));
// make sure incremental repair doesn't change anything;
CLUSTER.get(1).nodetool("repair", KEYSPACE);
CLUSTER.forEach(inst -> inst.runOnInstance(() -> {
Map<String, Long> sizes = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopSizePartitions();
for (int i = 99; i >= 90; i--)
assertTrue(sizes.containsKey(String.valueOf(i)));
}));
}
@Test
public void configChangeTest()
{
String name = "tbl" + COUNTER.getAndIncrement();
String table = KEYSPACE + "." + name;
CLUSTER.schemaChange("create table " + table + " (id int, ck int, t int, primary key (id, ck))");
for (int i = 0; i < 100; i++)
{
for (int j = 0; j < i; j++)
{
CLUSTER.coordinator(1).execute("insert into " + table + " (id, ck, t) values (?,?,?)", ConsistencyLevel.ALL, i, j, i * j + 100);
CLUSTER.coordinator(1).execute("DELETE FROM " + table + " where id = ? and ck = ?", ConsistencyLevel.ALL, i, -j);
}
}
// top should have 10 elements
repair();
CLUSTER.get(1).runOnInstance(() -> {
ColumnFamilyStore store = Keyspace.open(KEYSPACE).getColumnFamilyStore(name);
Assertions.assertThat(store.getTopTombstonePartitions()).hasSize(10);
Assertions.assertThat(store.getTopTombstonePartitions()).hasSize(10);
});
// reconfigure and repair; top should have 20 elements
setCount(20, 20);
repair();
CLUSTER.get(1).runOnInstance(() -> {
ColumnFamilyStore store = Keyspace.open(KEYSPACE).getColumnFamilyStore(name);
Assertions.assertThat(store.getTopTombstonePartitions()).hasSize(20);
Assertions.assertThat(store.getTopTombstonePartitions()).hasSize(20);
});
// test shrinking config
setCount(5, 5);
repair();
CLUSTER.get(1).runOnInstance(() -> {
ColumnFamilyStore store = Keyspace.open(KEYSPACE).getColumnFamilyStore(name);
Assertions.assertThat(store.getTopTombstonePartitions()).hasSize(5);
Assertions.assertThat(store.getTopTombstonePartitions()).hasSize(5);
});
}
@Test
public void basicRowTombstonesTest() throws InterruptedException
{
String name = "tbl" + COUNTER.getAndIncrement();
String table = KEYSPACE + "." + name;
CLUSTER.schemaChange("create table " + table + " (id int, ck int, t int, primary key (id, ck)) with gc_grace_seconds = 1");
for (int i = 0; i < 100; i++)
for (int j = 0; j < i; j++)
CLUSTER.coordinator(1).execute("DELETE FROM " + table + " where id = ? and ck = ?", ConsistencyLevel.ALL, i, j);
repair();
// tombstones not purgeable
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
for (int i = 99; i >= 90; i--)
{
assertTrue(tombstones.containsKey(String.valueOf(i)));
assertEquals(i, (long) tombstones.get(String.valueOf(i)));
}
});
Thread.sleep(2000);
// count purgeable tombstones;
repair();
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
for (int i = 99; i >= 90; i--)
{
assertTrue(tombstones.containsKey(String.valueOf(i)));
assertEquals(i, (long) tombstones.get(String.valueOf(i)));
}
});
CLUSTER.get(1).forceCompact(KEYSPACE, name);
// all tombstones actually purged;
repair();
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
assertTrue(tombstones.values().stream().allMatch(l -> l == 0));
});
}
@Test
public void basicRegularTombstonesTest() throws InterruptedException
{
String name = "tbl" + COUNTER.getAndIncrement();
String table = KEYSPACE + "." + name;
CLUSTER.schemaChange("create table " + table + " (id int, ck int, t int, primary key (id, ck)) with gc_grace_seconds = 1");
for (int i = 0; i < 100; i++)
for (int j = 0; j < i; j++)
CLUSTER.coordinator(1).execute("UPDATE " + table + " SET t = null where id = ? and ck = ?", ConsistencyLevel.ALL, i, j);
repair();
// tombstones not purgeable
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
for (int i = 99; i >= 90; i--)
{
assertTrue(tombstones.containsKey(String.valueOf(i)));
assertEquals(i, (long) tombstones.get(String.valueOf(i)));
}
});
Thread.sleep(2000);
// count purgeable tombstones;
repair();
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
for (int i = 99; i >= 90; i--)
{
assertTrue(tombstones.containsKey(String.valueOf(i)));
assertEquals(i, (long) tombstones.get(String.valueOf(i)));
}
});
CLUSTER.get(1).forceCompact(KEYSPACE, name);
// all tombstones actually purged;
repair();
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
assertTrue(tombstones.values().stream().allMatch(l -> l == 0));
});
}
private static void setCount(int size, int tombstone)
{
CLUSTER.forEach(i -> i.runOnInstance(() -> {
DatabaseDescriptor.setMaxTopSizePartitionCount(size);
DatabaseDescriptor.setMaxTopTombstonePartitionCount(tombstone);
}));
}
private void repair()
{
switch (repair)
{
case Incremental:
{
// IR will not populate, as it only looks at non-repaired data
// to trigger this patch, we need IR + --validate
CLUSTER.get(1).nodetoolResult("repair", KEYSPACE).asserts().success();
CLUSTER.get(1).nodetoolResult("repair", "--validate", KEYSPACE).asserts().success();
}
break;
case Full:
{
CLUSTER.get(1).nodetoolResult("repair", "-full", KEYSPACE).asserts().success();
}
break;
case FullPreview:
{
CLUSTER.get(1).nodetoolResult("repair", "-full", "--preview", KEYSPACE).asserts().success();
}
break;
default:
throw new AssertionError("Unknown repair type: " + repair);
}
}
@Test
public void basicRangeTombstonesTest() throws Throwable
{
String name = "tbl" + COUNTER.getAndIncrement();
String table = KEYSPACE + "." + name;
CLUSTER.schemaChange("create table " + table + " (id int, ck int, t int, primary key (id, ck)) with gc_grace_seconds = 1");
for (int i = 0; i < 100; i++)
for (int j = 0; j < i; j++)
CLUSTER.coordinator(1).execute("DELETE FROM " + table + " WHERE id = ? and ck >= ? and ck <= ?", ConsistencyLevel.ALL, i, j, j);
repair();
// tombstones not purgeable
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
// note that we count range tombstone markers - so the count will be double the number of deletions we did above
for (int i = 99; i >= 90; i--)
assertEquals(i * 2, (long)tombstones.get(String.valueOf(i)));
});
Thread.sleep(2000);
// count purgeable tombstones;
repair();
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
for (int i = 99; i >= 90; i--)
assertEquals(i * 2, (long)tombstones.get(String.valueOf(i)));
});
CLUSTER.get(1).forceCompact(KEYSPACE, name);
// all tombstones actually purged;
repair();
CLUSTER.get(1).runOnInstance(() -> {
Map<String, Long> tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions();
assertTrue(tombstones.values().stream().allMatch( l -> l == 0));
});
}
}

334
test/unit/org/apache/cassandra/db/TopPartitionTrackerTest.java Normal file
View File

@ -0,0 +1,334 @@
/*
* 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;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.Set;
import org.junit.Test;
import org.apache.cassandra.config.DataStorageSpec;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.metrics.TopPartitionTracker;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Pair;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class TopPartitionTrackerTest extends CQLTester
{
@Test
public void testSizeLimit()
{
createTable("create table %s (id bigint primary key, x int)");
DatabaseDescriptor.setMaxTopSizePartitionCount(5);
DatabaseDescriptor.setMinTrackedPartitionSize(new DataStorageSpec("12B"));
Collection<Range<Token>> fullRange = Collections.singleton(r(0, 0));
TopPartitionTracker tpt = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata());
TopPartitionTracker.Collector collector = new TopPartitionTracker.Collector(fullRange);
for (int i = 5; i < 15; i++)
collector.trackPartitionSize(dk(i), i);
tpt.merge(collector);
assertEquals(3, tpt.topSizes().top.size());
assertTrue(tpt.topSizes().top.stream().allMatch(tp -> tp.value >= 12));
Collection<Range<Token>> keyRange = rangesFor(7);
collector = new TopPartitionTracker.Collector(keyRange);
collector.trackPartitionSize(dk(7), 7);
tpt.merge(collector);
assertEquals(3, tpt.topSizes().top.size());
assertTrue(tpt.topSizes().top.stream().allMatch(tp -> tp.value >= 12));
assertFalse(tpt.topSizes().top.contains(tp(7, 7)));
collector = new TopPartitionTracker.Collector(keyRange);
collector.trackPartitionSize(dk(7), 100);
tpt.merge(collector);
assertEquals(4, tpt.topSizes().top.size());
assertTrue(tpt.topSizes().top.stream().allMatch(tp -> tp.value >= 12));
assertTrue(tpt.topSizes().top.contains(tp(7, 100)));
}
@Test
public void testCountLimit()
{
createTable("create table %s (id bigint primary key, x int)");
DatabaseDescriptor.setMinTrackedPartitionSize(new DataStorageSpec("0B"));
DatabaseDescriptor.setMaxTopSizePartitionCount(5);
Collection<Range<Token>> fullRange = Collections.singleton(r(0, 0));
TopPartitionTracker tpt = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata());
TopPartitionTracker.Collector collector = new TopPartitionTracker.Collector(fullRange);
for (int i = 5; i < 15; i++)
collector.trackPartitionSize(dk(i), i);
tpt.merge(collector);
assertEquals(5, tpt.topSizes().top.size());
collector = new TopPartitionTracker.Collector(fullRange);
for (int i = 5; i < 15; i++)
collector.trackPartitionSize(dk(i), i + 1);
tpt.merge(collector);
assertEquals(5, tpt.topSizes().top.size());
collector = new TopPartitionTracker.Collector(rangesFor(15));
collector.trackPartitionSize(dk(15), 14);
tpt.merge(collector);
assertEquals(5, tpt.topSizes().top.size());
}
@Test
public void testSubRangeMerge()
{
createTable("create table %s (id bigint primary key, x int)");
DatabaseDescriptor.setMinTrackedPartitionSize(new DataStorageSpec("0B"));
DatabaseDescriptor.setMaxTopSizePartitionCount(10);
Collection<Range<Token>> fullRange = Collections.singleton(r(0, 0));
TopPartitionTracker tpt = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata());
TopPartitionTracker.Collector collector = new TopPartitionTracker.Collector(fullRange);
for (int i = 0; i < 10; i++)
collector.trackPartitionSize(dk(i), 10);
tpt.merge(collector);
assertEquals(10, tpt.topSizes().top.size());
collector = new TopPartitionTracker.Collector(rangesFor(0,1,2,3,4));
for (int i = 0; i < 5; i++)
collector.trackPartitionSize(dk(i), 8);
tpt.merge(collector);
assertEquals(10, tpt.topSizes().top.size());
for (TopPartitionTracker.TopPartition tp : tpt.topSizes().top)
{
long key = ByteBufferUtil.toLong(tp.key.getKey());
if (key < 5)
assertEquals(8, tp.value);
else
assertEquals(10, tp.value);
}
}
@Test
public void testSaveLoad()
{
createTable("create table %s (id bigint primary key, x int)");
DatabaseDescriptor.setMinTrackedPartitionSize(new DataStorageSpec("0B"));
DatabaseDescriptor.setMinTrackedPartitionTombstoneCount(0);
DatabaseDescriptor.setMaxTopSizePartitionCount(10);
DatabaseDescriptor.setMaxTopTombstonePartitionCount(10);
Collection<Range<Token>> fullRange = Collections.singleton(r(0, 0));
TopPartitionTracker tpt = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata());
assertEquals(0, tpt.topSizes().lastUpdate);
assertEquals(0, tpt.topTombstones().lastUpdate);
long start = System.currentTimeMillis();
TopPartitionTracker.Collector collector = new TopPartitionTracker.Collector(fullRange);
for (int i = 0; i < 10; i++)
{
collector.trackPartitionSize(dk(i), 10);
collector.trackTombstoneCount(dk(i + 10), 100);
}
tpt.merge(collector);
long sizeUpdate = tpt.topSizes().lastUpdate;
long tombstoneUpdate = tpt.topTombstones().lastUpdate;
assertTrue(sizeUpdate >= start && sizeUpdate <= System.currentTimeMillis());
assertTrue(tombstoneUpdate >= start && tombstoneUpdate <= System.currentTimeMillis());
assertEquals(10, tpt.topSizes().top.size());
assertEquals(10, tpt.topTombstones().top.size());
tpt.save();
TopPartitionTracker tptLoaded = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata());
assertEquals(sizeUpdate, tptLoaded.topSizes().lastUpdate);
assertEquals(tombstoneUpdate, tptLoaded.topTombstones().lastUpdate);
assertEquals(tpt.topSizes().top, tptLoaded.topSizes().top);
assertEquals(tpt.topTombstones().top, tptLoaded.topTombstones().top);
DatabaseDescriptor.setMaxTopSizePartitionCount(5);
DatabaseDescriptor.setMaxTopTombstonePartitionCount(5);
tptLoaded = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata());
assertEquals(5, tptLoaded.topSizes().top.size());
assertEquals(5, tptLoaded.topTombstones().top.size());
assertEquals(sizeUpdate, tptLoaded.topSizes().lastUpdate);
assertEquals(tombstoneUpdate, tptLoaded.topTombstones().lastUpdate);
Iterator<TopPartitionTracker.TopPartition> oldIter = tpt.topSizes().top.iterator();
Iterator<TopPartitionTracker.TopPartition> loadedIter = tptLoaded.topSizes().top.iterator();
while (loadedIter.hasNext())
{
TopPartitionTracker.TopPartition old = oldIter.next();
TopPartitionTracker.TopPartition loaded = loadedIter.next();
assertEquals(old.key, loaded.key);
assertEquals(old.value, loaded.value);
}
oldIter = tpt.topTombstones().top.iterator();
loadedIter = tptLoaded.topTombstones().top.iterator();
while (loadedIter.hasNext())
{
TopPartitionTracker.TopPartition old = oldIter.next();
TopPartitionTracker.TopPartition loaded = loadedIter.next();
assertEquals(old.key, loaded.key);
assertEquals(old.value, loaded.value);
}
}
@Test
public void randomTest()
{
createTable("create table %s (id bigint primary key, x int)");
DatabaseDescriptor.setMinTrackedPartitionSize(new DataStorageSpec("0B"));
DatabaseDescriptor.setMaxTopSizePartitionCount(1000);
int keyCount = 10000;
long seed = System.currentTimeMillis();
Random r = new Random(seed);
List<DecoratedKey> keys = new ArrayList<>(keyCount);
for (int i = 0; i < keyCount; i++)
keys.add(dk(i));
Collection<Range<Token>> fullRange = Collections.singleton(r(0, 0));
List<Pair<DecoratedKey, Long>> expected = new ArrayList<>();
TopPartitionTracker tpt = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata());
TopPartitionTracker.Collector collector = new TopPartitionTracker.Collector(fullRange);
Set<Long> uniqueValues = new HashSet<>();
for (int i = 0; i < keys.size(); i++)
{
DecoratedKey key = keys.get(i);
long value;
do
{
value = Math.abs(r.nextLong() % 100000);
} while (!uniqueValues.add(value));
expected.add(Pair.create(key, value));
collector.trackPartitionSize(key, value);
}
assertEquals(keyCount, expected.size());
tpt.merge(collector);
expected.sort((o1, o2) -> {
int cmp = -o1.right.compareTo(o2.right);
if (cmp != 0)
return cmp;
return o1.left.compareTo(o2.left);
});
Iterator<Pair<DecoratedKey, Long>> expectedTop = expected.subList(0,1000).iterator();
Iterator<TopPartitionTracker.TopPartition> trackedTop = tpt.topSizes().top.iterator();
while (expectedTop.hasNext())
{
Pair<DecoratedKey, Long> ex = expectedTop.next();
TopPartitionTracker.TopPartition tracked = trackedTop.next();
assertEquals("seed "+seed, ex.left, tracked.key);
assertEquals("seed "+seed, (long)ex.right, tracked.value);
}
}
@Test
public void testRanges() throws UnknownHostException
{
createTable("create table %s (id bigint primary key, x int)");
DatabaseDescriptor.setMinTrackedPartitionSize(new DataStorageSpec("0B"));
DatabaseDescriptor.setMaxTopSizePartitionCount(1000);
long seed = System.currentTimeMillis();
Random r = new Random(seed);
List<Pair<DecoratedKey, Long>> keys = new ArrayList<>(10000);
for (int i = 0; i < 10000; i++)
keys.add(Pair.create(dk(i), Math.abs(r.nextLong() % 20000)));
Collection<Range<Token>> fullRange = Collections.singleton(r(0, 0));
TopPartitionTracker tpt = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata());
TopPartitionTracker.Collector collector = new TopPartitionTracker.Collector(fullRange);
for (int i = 0; i < keys.size(); i++)
{
Pair<DecoratedKey, Long> entry = keys.get(i);
collector.trackPartitionSize(entry.left, entry.right);
}
tpt.merge(collector);
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
tmd.updateNormalToken(t(0), InetAddressAndPort.getByName("127.0.0.1"));
tmd.updateNormalToken(t(Long.MAX_VALUE - 1), InetAddressAndPort.getByName("127.0.0.2"));
Iterator<TopPartitionTracker.TopPartition> trackedTop = tpt.topSizes().top.iterator();
Collection<Range<Token>> localRanges = StorageService.instance.getLocalReplicas(keyspace()).ranges();
int outOfRangeCount = 0;
while (trackedTop.hasNext())
{
if (!Range.isInRanges(trackedTop.next().key.getToken(), localRanges))
outOfRangeCount++;
}
assertTrue(outOfRangeCount > 0);
collector = new TopPartitionTracker.Collector(localRanges);
for (int i = 0; i < keys.size(); i++)
{
Pair<DecoratedKey, Long> entry = keys.get(i);
// we don't need this check during compaction since we know we won't track any tokens outside the owned ranges
// but the TopPartitionTracker might still be tracking outside of the local ranges - these are cleared in .merge()
if (Range.isInRanges(entry.left.getToken(), localRanges))
collector.trackPartitionSize(entry.left, entry.right);
}
tpt.merge(collector);
outOfRangeCount = 0;
trackedTop = tpt.topSizes().top.iterator();
while (trackedTop.hasNext())
{
if (!Range.isInRanges(trackedTop.next().key.getToken(), localRanges))
outOfRangeCount++;
}
assertEquals(0, outOfRangeCount);
assertTrue(tpt.topSizes().top.size() > 0);
}
private static TopPartitionTracker.TopPartition tp(int i, long c)
{
return new TopPartitionTracker.TopPartition(dk(i), c);
}
private static DecoratedKey dk(long i)
{
return Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(i));
}
private static Range<Token> r(long start, long end)
{
return new Range<>(t(start), t(end));
}
private static Token t(long v)
{
return new Murmur3Partitioner.LongToken(v);
}
private static long tokenValue(long key)
{
return (long) dk(key).getToken().getTokenValue();
}
private static Collection<Range<Token>> rangesFor(long ... keys)
{
List<Range<Token>> ranges = new ArrayList<>(keys.length);
for (long key : keys)
ranges.add(r(tokenValue(key) - 1, tokenValue(key)));
return ranges;
}
}

2
test/unit/org/apache/cassandra/repair/ValidatorTest.java
View File

@ -114,7 +114,7 @@ public class ValidatorTest
validator.state.phase.start(10, 10);
MerkleTrees trees = new MerkleTrees(partitioner);
trees.addMerkleTrees((int) Math.pow(2, 15), validator.desc.ranges);
validator.prepare(cfs, trees);
validator.prepare(cfs, trees, null);
// and confirm that the trees were split
assertTrue(trees.size() > 1);