diff --git a/NEWS.txt b/NEWS.txt index fd31e06c93..72c95e00c4 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -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 diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index e0961ed945..2f339418b1 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -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 getOverrideLoadConfig() { return overrideLoadConfig; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index c1fee6e5c8..0ce787fe32 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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; + } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 47dd66d7ae..5db768ca6e 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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 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 getTopTombstonePartitions() + { + if (topPartitions == null) + return Collections.emptyMap(); + return topPartitions.getTopTombstonePartitionMap(); + } + + @Override + public Long getTopTombstonePartitionsLastUpdate() + { + if (topPartitions == null) + return null; + return topPartitions.topTombstones().lastUpdate; + } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java index dd68c522fc..5b6fd16fe1 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java @@ -266,4 +266,9 @@ public interface ColumnFamilyStoreMBean public boolean hasMisplacedSSTables(); public List getDataPaths() throws IOException; + + public Map getTopSizePartitions(); + public Long getTopSizePartitionsLastUpdate(); + public Map getTopTombstonePartitions(); + public Long getTopTombstonePartitionsLastUpdate(); } diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 8497dad25f..a4d71fe132 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -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>>," + + "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 { D accept(A var1, B var2, C var3); } + + public static void saveTopPartitions(TableMetadata metadata, String topType, Collection 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 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 top = row.getList("top", BytesType.instance); + if (top == null || top.isEmpty()) + return TopPartitionTracker.StoredTopPartitions.EMPTY; + + List 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; + } + } } diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 17094629c5..a0dc0875d7 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -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 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 scanners, AbstractCompactionController controller, int nowInSec, TimeUUID compactionId, ActiveCompactionsTracker activeCompactions) + public CompactionIterator(OperationType type, + List 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 purger = isPaxos(controller.cfs) && paxosStatePurging() != legacy ? new PaxosPurger(nowInSec) diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 165e1e02f3..0900a3ffab 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -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 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 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(); diff --git a/src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java b/src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java index d656b04529..0f614dd901 100644 --- a/src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java +++ b/src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java @@ -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> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec) throws IOException, NoSuchRepairSessionException + public ValidationPartitionIterator getValidationIterator(Collection> 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 diff --git a/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java b/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java index 1eecf6623e..023d2344b8 100644 --- a/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java +++ b/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java @@ -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 scanners, ValidationCompactionController controller, int nowInSec, ActiveCompactionsTracker activeCompactions) + public ValidationCompactionIterator(List 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, Long> rangePartitionCounts; - public CassandraValidationIterator(ColumnFamilyStore cfs, Collection> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec) throws IOException, NoSuchRepairSessionException + public CassandraValidationIterator(ColumnFamilyStore cfs, Collection> 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()); diff --git a/src/java/org/apache/cassandra/metrics/TopPartitionTracker.java b/src/java/org/apache/cassandra/metrics/TopPartitionTracker.java new file mode 100644 index 0000000000..f954b58bf9 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/TopPartitionTracker.java @@ -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 topSizes = new AtomicReference<>(); + private final AtomicReference 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 getTopTombstonePartitionMap() + { + return topTombstones.get().toMap(metadata); + } + + public Map 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> 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 top; + private final int maxTopPartitionCount; + private final long minTrackedValue; + private final Collection> ranges; + private long currentMinValue = Long.MAX_VALUE; + public final long lastUpdate; + + private TopHolder(int maxTopPartitionCount, long minTrackedValue, Collection> ranges) + { + this(maxTopPartitionCount, minTrackedValue, new TreeSet<>(), ranges, 0); + } + + private TopHolder(int maxTopPartitionCount, long minTrackedValue, NavigableSet top, Collection> 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> 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 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 toMap(TableMetadata metadata) + { + Map 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 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 + { + 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 + { + 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 topPartitions; + public final long lastUpdated; + + public StoredTopPartitions(List topPartitions, long lastUpdated) + { + this.topPartitions = topPartitions; + this.lastUpdated = lastUpdated; + } + } +} diff --git a/src/java/org/apache/cassandra/repair/TableRepairManager.java b/src/java/org/apache/cassandra/repair/TableRepairManager.java index 0210afdb7a..622374ae99 100644 --- a/src/java/org/apache/cassandra/repair/TableRepairManager.java +++ b/src/java/org/apache/cassandra/repair/TableRepairManager.java @@ -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> ranges, TimeUUID parentId, TimeUUID sessionID, boolean isIncremental, int nowInSec) throws IOException, NoSuchRepairSessionException; + ValidationPartitionIterator getValidationIterator(Collection> 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. diff --git a/src/java/org/apache/cassandra/repair/ValidationManager.java b/src/java/org/apache/cassandra/repair/ValidationManager.java index ccfccea793..a6c9fa9ef2 100644 --- a/src/java/org/apache/cassandra/repair/ValidationManager.java +++ b/src/java/org/apache/cassandra/repair/ValidationManager.java @@ -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. */ diff --git a/src/java/org/apache/cassandra/repair/Validator.java b/src/java/org/apache/cassandra/repair/Validator.java index 8b7f52dbc9..abbfb4f9c5 100644 --- a/src/java/org/apache/cassandra/repair/Validator.java +++ b/src/java/org/apache/cassandra/repair/Validator.java @@ -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); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index c24096ad12..156f54715b 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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); + } } diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 802295924b..dd8f93fd31 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -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); } diff --git a/src/java/org/apache/cassandra/tools/nodetool/stats/StatsTable.java b/src/java/org/apache/cassandra/tools/nodetool/stats/StatsTable.java index 139d9efd8d..8b5090bd86 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/stats/StatsTable.java +++ b/src/java/org/apache/cassandra/tools/nodetool/stats/StatsTable.java @@ -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 sstableBytesInEachLevel = new ArrayList<>(); public Boolean isInCorrectLocation = null; // null: option not active public double droppableTombstoneRatio; + public Map topSizePartitions; + public Map topTombstonePartitions; + public String topSizePartitionsLastUpdate; + public String topTombstonePartitionsLastUpdate; } diff --git a/src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsHolder.java b/src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsHolder.java index 9107007439..c6b2301c2c 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsHolder.java +++ b/src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsHolder.java @@ -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 format(Map map, boolean humanReadable) + { + LinkedHashMap retMap = new LinkedHashMap<>(); + for (Map.Entry 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. */ diff --git a/src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsPrinter.java b/src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsPrinter.java index ac10d7c737..7f17c21d6d 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsPrinter.java +++ b/src/java/org/apache/cassandra/tools/nodetool/stats/TableStatsPrinter.java @@ -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 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 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 tombstonecnt : table.topTombstonePartitions.entrySet()) + out.printf(indent + " %-" + maxWidth + "s %s%n", tombstonecnt.getKey(), tombstonecnt.getValue()); + } out.println(""); } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/TopPartitionsTest.java b/test/distributed/org/apache/cassandra/distributed/test/TopPartitionsTest.java new file mode 100644 index 0000000000..335483454b --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/TopPartitionsTest.java @@ -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 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 sizes = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopSizePartitions(); + for (int i = 99; i >= 90; i--) + assertTrue(sizes.containsKey(String.valueOf(i))); + + Map 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 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 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 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 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 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 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 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 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 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 tombstones = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).getTopTombstonePartitions(); + assertTrue(tombstones.values().stream().allMatch( l -> l == 0)); + }); + } +} diff --git a/test/unit/org/apache/cassandra/db/TopPartitionTrackerTest.java b/test/unit/org/apache/cassandra/db/TopPartitionTrackerTest.java new file mode 100644 index 0000000000..938be11818 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/TopPartitionTrackerTest.java @@ -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> 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> 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> 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> 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> 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 oldIter = tpt.topSizes().top.iterator(); + Iterator 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 keys = new ArrayList<>(keyCount); + for (int i = 0; i < keyCount; i++) + keys.add(dk(i)); + + Collection> fullRange = Collections.singleton(r(0, 0)); + List> expected = new ArrayList<>(); + TopPartitionTracker tpt = new TopPartitionTracker(getCurrentColumnFamilyStore().metadata()); + TopPartitionTracker.Collector collector = new TopPartitionTracker.Collector(fullRange); + Set 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> expectedTop = expected.subList(0,1000).iterator(); + Iterator trackedTop = tpt.topSizes().top.iterator(); + + while (expectedTop.hasNext()) + { + Pair 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> keys = new ArrayList<>(10000); + for (int i = 0; i < 10000; i++) + keys.add(Pair.create(dk(i), Math.abs(r.nextLong() % 20000))); + + Collection> 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 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 trackedTop = tpt.topSizes().top.iterator(); + Collection> 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 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 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> rangesFor(long ... keys) + { + List> ranges = new ArrayList<>(keys.length); + for (long key : keys) + ranges.add(r(tokenValue(key) - 1, tokenValue(key))); + return ranges; + } +} diff --git a/test/unit/org/apache/cassandra/repair/ValidatorTest.java b/test/unit/org/apache/cassandra/repair/ValidatorTest.java index a5d8aebb0a..4c78b743ad 100644 --- a/test/unit/org/apache/cassandra/repair/ValidatorTest.java +++ b/test/unit/org/apache/cassandra/repair/ValidatorTest.java @@ -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);