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

Improved observability in AutoRepair to report both expected vs. actual repair bytes and expected vs. actual keyspaces 

patch by Jaydeepkumar Chovatia; reviewed by Chris Lohfink for CASSANDRA-20581
This commit is contained in:
jaydeepkumar1984 2025-04-27 10:47:32 -07:00 committed by Jaydeepkumar Chovatia
parent d7a46b52ef
commit bf2c1c124e
20 changed files with 1170 additions and 405 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Improved observability in AutoRepair to report both expected vs. actual repair bytes and expected vs. actual keyspaces (CASSANDRA-20581)
* Execution of CreateTriggerStatement should not rely on external state (CASSANDRA-20287)
* Support LIKE expressions in filtering queries (CASSANDRA-17198)
* Make legacy index rebuilds safe on Gossip -> TCM upgrades (CASSANDRA-20887)

42
doc/modules/cassandra/pages/managing/operating/metrics.adoc
View File

@ -1085,40 +1085,58 @@ Reported name format:
|===
|Name |Type |Description
|RepairsInProgress |Gauge<Integer> |Repair is in progress
on the node
on the node.
|NodeRepairTimeInSec |Gauge<Integer> |Time taken to repair
the node in seconds
the node in seconds.
|ClusterRepairTimeInSec |Gauge<Integer> |Time taken to repair
the entire Cassandra cluster in seconds
the entire Cassandra cluster in seconds.
|LongestUnrepairedSec |Gauge<Integer> |Time since the last repair
ran on the node in seconds
ran on the node in seconds.
|RepairStartLagSec|Gauge<Integer> |If a repair has not run within min_repair_interval, how long past this value since
repairs last completed. Useful for determining if repairs are behind schedule.
|SucceededTokenRangesCount |Gauge<Integer> |Number of token ranges successfully repaired on the node
|SucceededTokenRangesCount |Gauge<Integer> |Number of token ranges successfully repaired on the node.
|FailedTokenRangesCount |Gauge<Integer> |Number of token ranges failed to repair on the node
|FailedTokenRangesCount |Gauge<Integer> |Number of token ranges failed to repair on the node.
|SkippedTokenRangesCount |Gauge<Integer> |Number of token ranges skipped
on the node
on the node.
|SkippedTablesCount |Gauge<Integer> |Number of tables skipped
on the node
on the node.
|TotalMVTablesConsideredForRepair |Gauge<Integer> |Number of materialized
views considered on the node
views considered on the node.
|TotalDisabledRepairTables |Gauge<Integer> |Number of tables on which
the automated repair has been disabled on the node
the automated repair has been disabled on the node.
|RepairTurnMyTurn |Counter |Represents the node's turn to repair
|TotalBytesToRepair |Gauge<Long> |Total bytes to be repaired across all keyspaces and tables involved in the current
repair schedule.
|BytesAlreadyRepaired |Gauge<Long> |Cumulative number of bytes successfully repaired so far in the current
repair schedule.
NOTE: This calculation is the best effort for the FixedSplitTokenRangeSplitter. In practice, this metric
may not give you an accurate view in case of uneven data distribution.
|TotalKeyspaceRepairPlansToRepair |Gauge<Integer> |Represents the total number of keyspace-level repair plans scheduled
for execution. If no table-level repair priorities are configured, this number typically matches the total number
of keyspaces under repair. However, if certain tables have repair priorities set, this number is usually higher than
the number of keyspaces, as multiple repair plans may be generated for different prioritized tables within the
same keyspace.
|KeyspaceRepairPlansAlreadyRepaired |Gauge<Integer> |Cumulative number of keyspace-level repair plans successfully
repaired so far in the current repair schedule.
|RepairTurnMyTurn |Counter |Represents the node's turn to repair.
|RepairTurnMyTurnDueToPriority |Counter |Represents the node's turn to repair
due to priority set in the automated repair
due to priority set in the automated repair.
|RepairDelayedByReplica |Counter |Represents occurrences of a node's turn being
delayed because a replica was currently taking its turn. Only relevant if

33
src/java/org/apache/cassandra/metrics/AutoRepairMetrics.java
View File

@ -46,6 +46,11 @@ public class AutoRepairMetrics
public final Gauge<Integer> skippedTablesCount;
public final Gauge<Integer> totalMVTablesConsideredForRepair;
public final Gauge<Integer> totalDisabledRepairTables;
public final Gauge<Long> totalBytesToRepair;
public final Gauge<Long> bytesAlreadyRepaired;
public final Gauge<Integer> totalKeyspaceRepairPlansToRepair;
public final Gauge<Integer> keyspaceRepairPlansAlreadyRepaired;
public Counter repairTurnMyTurn;
public Counter repairTurnMyTurnDueToPriority;
public Counter repairTurnMyTurnForceRepair;
@ -155,6 +160,34 @@ public class AutoRepairMetrics
return AutoRepair.instance.getRepairState(repairType).getTotalDisabledTablesRepairCount();
}
});
totalBytesToRepair = Metrics.register(factory.createMetricName("TotalBytesToRepair"), new Gauge<Long>()
{
public Long getValue()
{
return AutoRepair.instance.getRepairState(repairType).getTotalBytesToRepair();
}
});
bytesAlreadyRepaired = Metrics.register(factory.createMetricName("BytesAlreadyRepaired"), new Gauge<Long>()
{
public Long getValue()
{
return AutoRepair.instance.getRepairState(repairType).getBytesAlreadyRepaired();
}
});
totalKeyspaceRepairPlansToRepair = Metrics.register(factory.createMetricName("TotalKeyspaceRepairPlansToRepair"), new Gauge<Integer>()
{
public Integer getValue()
{
return AutoRepair.instance.getRepairState(repairType).getTotalKeyspaceRepairPlansToRepair();
}
});
keyspaceRepairPlansAlreadyRepaired = Metrics.register(factory.createMetricName("KeyspaceRepairPlansAlreadyRepaired"), new Gauge<Integer>()
{
public Integer getValue()
{
return AutoRepair.instance.getRepairState(repairType).getKeyspaceRepairPlansAlreadyRepaired();
}
});
}
public void recordTurn(AutoRepairUtils.RepairTurn turn)

26
src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java
View File

@ -123,14 +123,15 @@ public class AutoRepair
repairExecutors = new EnumMap<>(AutoRepairConfig.RepairType.class);
repairRunnableExecutors = new EnumMap<>(AutoRepairConfig.RepairType.class);
repairStates = new EnumMap<>(AutoRepairConfig.RepairType.class);
AutoRepairConfig config = DatabaseDescriptor.getAutoRepairConfig();
for (AutoRepairConfig.RepairType repairType : AutoRepairConfig.RepairType.values())
{
repairExecutors.put(repairType, executorFactory().scheduled(false, "AutoRepair-Repair-" + repairType.getConfigName(), Thread.NORM_PRIORITY));
repairRunnableExecutors.put(repairType, executorFactory().scheduled(false, "AutoRepair-RepairRunnable-" + repairType.getConfigName(), Thread.NORM_PRIORITY));
repairStates.put(repairType, AutoRepairConfig.RepairType.getAutoRepairState(repairType));
repairStates.put(repairType, AutoRepairConfig.RepairType.getAutoRepairState(repairType, config));
}
AutoRepairConfig config = DatabaseDescriptor.getAutoRepairConfig();
AutoRepairUtils.setup();
for (AutoRepairConfig.RepairType repairType : AutoRepairConfig.RepairType.values())
@ -197,6 +198,8 @@ public class AutoRepair
if (turn == MY_TURN || turn == MY_TURN_DUE_TO_PRIORITY || turn == MY_TURN_FORCE_REPAIR)
{
repairState.recordTurn(turn);
repairState.setBytesAlreadyRepaired(0L);
repairState.setKeyspaceRepairPlansAlreadyRepaired(0);
// For normal auto repair, we will use primary range only repairs (Repair with -pr option).
// For some cases, we may set the auto_repair_primary_token_range_only flag to false then we will do repair
// without -pr. We may also do force repair for certain node that we want to repair all the data on one node
@ -231,23 +234,30 @@ public class AutoRepair
}
// Separate out the keyspaces and tables to repair based on their priority, with each repair plan representing a uniquely occuring priority.
List<PrioritizedRepairPlan> repairPlans = PrioritizedRepairPlan.build(keyspacesAndTablesToRepair, repairType, shuffleFunc);
List<PrioritizedRepairPlan> repairPlans = PrioritizedRepairPlan.build(keyspacesAndTablesToRepair, repairType, shuffleFunc, primaryRangeOnly);
repairState.updateRepairScheduleStatistics(repairPlans);
// calculate the repair assignments for each priority:keyspace.
Iterator<KeyspaceRepairAssignments> repairAssignmentsIterator = config.getTokenRangeSplitterInstance(repairType).getRepairAssignments(primaryRangeOnly, repairPlans);
int keyspaceRepairAssignmentsAlreadyRepaired = 0;
while (repairAssignmentsIterator.hasNext())
{
KeyspaceRepairAssignments repairAssignments = repairAssignmentsIterator.next();
List<RepairAssignment> assignments = repairAssignments.getRepairAssignments();
if (assignments.isEmpty())
{
keyspaceRepairAssignmentsAlreadyRepaired++;
logger.info("Skipping repairs for priorityBucket={} for keyspace={} since it yielded no assignments", repairAssignments.getPriority(), repairAssignments.getKeyspaceName());
continue;
}
logger.info("Submitting repairs for priorityBucket={} for keyspace={} with assignmentCount={}", repairAssignments.getPriority(), repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments().size());
logger.info("Submitting repairs for priorityBucket={} for keyspace={} with assignmentCount={} and keyspaceRepairAssignmentsAlreadyRepaired={}/{}",
repairAssignments.getPriority(), repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments().size(),
keyspaceRepairAssignmentsAlreadyRepaired, repairState.getTotalKeyspaceRepairPlansToRepair());
repairKeyspace(repairType, primaryRangeOnly, repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments(), collectedRepairStats);
keyspaceRepairAssignmentsAlreadyRepaired++;
repairState.setKeyspaceRepairPlansAlreadyRepaired(keyspaceRepairAssignmentsAlreadyRepaired);
}
cleanupAndUpdateStats(turn, repairType, repairState, myId, startTimeInMillis, collectedRepairStats);
@ -277,6 +287,7 @@ public class AutoRepair
long tableStartTime = timeFunc.get();
int totalProcessedAssignments = 0;
Set<Range<Token>> ranges = new HashSet<>();
long bytesAlreadyRepaired = repairState.getBytesAlreadyRepaired();
for (RepairAssignment curRepairAssignment : repairAssignments)
{
try
@ -380,7 +391,10 @@ public class AutoRepair
}
ranges.clear();
}
logger.info("Repair completed for {} tables {}, range {}", keyspaceName, curRepairAssignment.getTableNames(), curRepairAssignment.getTokenRange());
bytesAlreadyRepaired += curRepairAssignment.getEstimatedBytes();
repairState.setBytesAlreadyRepaired(bytesAlreadyRepaired);
logger.info("Repair completed for {} tables {}, range {}, bytesAlreadyRepaired {}/{}",
keyspaceName, curRepairAssignment.getTableNames(), curRepairAssignment.getTokenRange(), bytesAlreadyRepaired, repairState.getTotalBytesToRepair());
}
catch (Exception e)
{
@ -492,8 +506,8 @@ public class AutoRepair
TimeUnit.SECONDS.toDays(repairState.getClusterRepairTimeInSec()));
}
repairState.setLastRepairTime(timeFunc.get());
repairState.setRepairInProgress(false);
AutoRepairUtils.updateFinishAutoRepairHistory(repairType, myId, timeFunc.get());
}

8
src/java/org/apache/cassandra/repair/autorepair/AutoRepairConfig.java
View File

@ -92,16 +92,16 @@ public class AutoRepairConfig implements Serializable
return configName;
}
public static AutoRepairState getAutoRepairState(RepairType repairType)
public static AutoRepairState getAutoRepairState(RepairType repairType, AutoRepairConfig config)
{
switch (repairType)
{
case FULL:
return new FullRepairState();
return new FullRepairState(config);
case INCREMENTAL:
return new IncrementalRepairState();
return new IncrementalRepairState(config);
case PREVIEW_REPAIRED:
return new PreviewRepairedState();
return new PreviewRepairedState(config);
}
throw new IllegalArgumentException("Invalid repair type: " + repairType);

77
src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java
View File

@ -60,6 +60,8 @@ public abstract class AutoRepairState
@VisibleForTesting
protected final RepairType repairType;
@VisibleForTesting
protected AutoRepairConfig config;
@VisibleForTesting
protected int totalTablesConsideredForRepair = 0;
@VisibleForTesting
protected long lastRepairTimeInMs;
@ -84,13 +86,22 @@ public abstract class AutoRepairState
@VisibleForTesting
protected int skippedTablesCount = 0;
@VisibleForTesting
protected long totalBytesToRepair = 0;
@VisibleForTesting
protected long bytesAlreadyRepaired = 0;
@VisibleForTesting
protected int totalKeyspaceRepairPlansToRepair = 0;
@VisibleForTesting
protected int keyspaceRepairPlansAlreadyRepaired = 0;
@VisibleForTesting
protected AutoRepairHistory longestUnrepairedNode;
protected final AutoRepairMetrics metrics;
protected AutoRepairState(RepairType repairType)
protected AutoRepairState(RepairType repairType, AutoRepairConfig config)
{
metrics = AutoRepairMetricsManager.getMetrics(repairType);
this.repairType = repairType;
this.config = config;
}
public abstract RepairCoordinator getRepairRunnable(String keyspace, List<String> tables, Set<Range<Token>> ranges, boolean primaryRangeOnly);
@ -98,7 +109,15 @@ public abstract class AutoRepairState
protected RepairCoordinator getRepairRunnable(String keyspace, RepairOption options)
{
return new RepairCoordinator(StorageService.instance, StorageService.nextRepairCommand.incrementAndGet(),
options, keyspace);
options, keyspace);
}
public void updateRepairScheduleStatistics(List<PrioritizedRepairPlan> repairPlans)
{
setTotalBytesToRepair(repairPlans.stream().
flatMap(repairPlan -> repairPlan.getKeyspaceRepairPlans().
stream()).mapToLong(KeyspaceRepairPlan::getEstimatedBytes).sum());
setTotalKeyspaceRepairPlansToRepair(repairPlans.stream().mapToInt(repairPlan -> repairPlan.getKeyspaceRepairPlans().size()).sum());
}
public long getLastRepairTime()
@ -239,20 +258,60 @@ public abstract class AutoRepairState
{
return totalDisabledTablesRepairCount;
}
public void setTotalBytesToRepair(long totalBytesToRepair)
{
this.totalBytesToRepair = totalBytesToRepair;
}
public long getTotalBytesToRepair()
{
return totalBytesToRepair;
}
public void setBytesAlreadyRepaired(long bytesAlreadyRepaired)
{
this.bytesAlreadyRepaired = bytesAlreadyRepaired;
}
public long getBytesAlreadyRepaired()
{
return bytesAlreadyRepaired;
}
public void setTotalKeyspaceRepairPlansToRepair(int totalKeyspaceRepairPlansToRepair)
{
this.totalKeyspaceRepairPlansToRepair = totalKeyspaceRepairPlansToRepair;
}
public int getTotalKeyspaceRepairPlansToRepair()
{
return totalKeyspaceRepairPlansToRepair;
}
public void setKeyspaceRepairPlansAlreadyRepaired(int keyspaceRepairPlansAlreadyRepaired)
{
this.keyspaceRepairPlansAlreadyRepaired = keyspaceRepairPlansAlreadyRepaired;
}
public int getKeyspaceRepairPlansAlreadyRepaired()
{
return keyspaceRepairPlansAlreadyRepaired;
}
}
class PreviewRepairedState extends AutoRepairState
{
public PreviewRepairedState()
public PreviewRepairedState(AutoRepairConfig config)
{
super(RepairType.PREVIEW_REPAIRED);
super(RepairType.PREVIEW_REPAIRED, config);
}
@Override
public RepairCoordinator getRepairRunnable(String keyspace, List<String> tables, Set<Range<Token>> ranges, boolean primaryRangeOnly)
{
RepairOption option = new RepairOption(RepairParallelism.PARALLEL, primaryRangeOnly, false, false,
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges, false, false, PreviewKind.REPAIRED, false, true, true, false, false, false);
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges, false, false, PreviewKind.REPAIRED, false, true, true, false, false, false);
option.getColumnFamilies().addAll(tables);
@ -262,9 +321,9 @@ class PreviewRepairedState extends AutoRepairState
class IncrementalRepairState extends AutoRepairState
{
public IncrementalRepairState()
public IncrementalRepairState(AutoRepairConfig config)
{
super(RepairType.INCREMENTAL);
super(RepairType.INCREMENTAL, config);
}
@Override
@ -307,9 +366,9 @@ class IncrementalRepairState extends AutoRepairState
class FullRepairState extends AutoRepairState
{
public FullRepairState()
public FullRepairState(AutoRepairConfig config)
{
super(RepairType.FULL);
super(RepairType.FULL, config);
}
@Override

223
src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.repair.autorepair;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@ -43,9 +44,15 @@ import com.google.common.base.MoreObjects;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.clearspring.analytics.stream.cardinality.CardinalityMergeException;
import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
import com.clearspring.analytics.stream.cardinality.ICardinality;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Splitter;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.metadata.CompactionMetadata;
import org.apache.cassandra.io.sstable.metadata.MetadataType;
import org.apache.cassandra.locator.EndpointsByRange;
import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.locator.LocalStrategy;
@ -83,6 +90,7 @@ import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
@ -93,6 +101,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.repair.autorepair.AutoRepairConfig.RepairType;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.RepairTurn.MY_TURN;
import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.RepairTurn.MY_TURN_DUE_TO_PRIORITY;
@ -312,7 +321,7 @@ public class AutoRepairUtils
public Set<UUID> getAllHostsWithOngoingRepair()
{
return Sets.union(hostIdsWithOnGoingRepair, hostIdsWithOnGoingForceRepair);
return Sets.union(hostIdsWithOnGoingRepair, hostIdsWithOnGoingForceRepair);
}
public String toString()
@ -484,6 +493,7 @@ public class AutoRepairUtils
/**
* Convenience method to resolve the broadcast address of a host id from {@link ClusterMetadata}
*
* @return broadcast address if it exists in CMS, otherwise null.
*/
@Nullable
@ -551,7 +561,7 @@ public class AutoRepairUtils
* Accepts the currently evaluated repairType's schedule as an optimization to avoid grabbing its repair status an
* additional time.
*
* @param myRepairType The repair type schedule being evaluated.
* @param myRepairType The repair type schedule being evaluated.
* @param myRepairStatus The repair status for that repair type.
* @return All hosts among active schedules currently being repaired.
*/
@ -591,6 +601,7 @@ public class AutoRepairUtils
* Identifies the most eligible host to repair for nodes preceding or equal to this nodes' lastRepairFinishTime.
* The criteria for this is to find the node with the oldest last repair finish time of which none of its replicas
* are currently under repair.
*
* @return The most eligible host to repair or null if no candidates before and including this nodes' current repair status.
*/
@VisibleForTesting
@ -666,13 +677,13 @@ public class AutoRepairUtils
/**
* @return Whether the host for the given eligibleRepairHistory has any replicas in hostsBeingRepaired.
* @param eligibleHistory History of node to check
* @param myId Host id of this node, if the repair history is for this node, additional logging will take place.
* @param myRepairType repair type being evaluated
* @param hostsBeingRepaired Hosts being repaired.
* @param hostIdToRepairType mapping of hosts being repaired to the repair type its being repaired for.
* @param eligibleHistory History of node to check
* @param myId Host id of this node, if the repair history is for this node, additional logging will take place.
* @param myRepairType repair type being evaluated
* @param hostsBeingRepaired Hosts being repaired.
* @param hostIdToRepairType mapping of hosts being repaired to the repair type its being repaired for.
* @param replicationStrategies Mapping of unique replication strategies to keyspaces having that strategy.
* @return Whether the host for the given eligibleRepairHistory has any replicas in hostsBeingRepaired.
*/
private static boolean hasReplicaWithOngoingRepair(AutoRepairHistory eligibleHistory,
UUID myId,
@ -726,7 +737,6 @@ public class AutoRepairUtils
eligibleHistory.hostId, eligibleBroadcastAddress,
hostId, inetAddressAndPort, entry.getValue().size(), entry.getValue().get(0),
hostIdToRepairType.get(hostId));
}
return true;
}
@ -871,7 +881,7 @@ public class AutoRepairUtils
}
int parallelRepairNumber = getMaxNumberOfNodeRunAutoRepair(repairType,
autoRepairHistories == null ? 0 : autoRepairHistories.size());
autoRepairHistories == null ? 0 : autoRepairHistories.size());
logger.info("Will run repairs concurrently on {} node(s)", parallelRepairNumber);
if (currentRepairStatus == null || parallelRepairNumber > currentRepairStatus.hostIdsWithOnGoingRepair.size())
{
@ -937,8 +947,8 @@ public class AutoRepairUtils
// log which node is next, which is helpful for debugging
logger.info("Next node to be repaired for repair type {}: {} ({})", repairType,
getBroadcastAddress(nodeToBeRepaired.hostId),
nodeToBeRepaired);
getBroadcastAddress(nodeToBeRepaired.hostId),
nodeToBeRepaired);
}
// If this node is not identified as most eligible, set the repair lag time.
@ -1201,4 +1211,193 @@ public class AutoRepairUtils
}
return ranges;
}
/**
* Finds a list of SSTables for a given {@code repairType},
* {@code keyspace}, {@code table}, and {@code tokenRange} and then it internally calls
* another API {@code AutoRepairUtils.getSizesForRangeOfSSTables}, which figures out the estimated data size.
*
* @param repairType the repair type (e.g., FULL, INCREMENTAL)
* @param keyspace the keyspace name
* @param table the table name
* @param tokenRange the token range to evaluate
* @return an estimate representing the number of partitions, size in range, and total size
*/
static SizeEstimate getRangeSizeEstimate(RepairType repairType, String keyspace, String table, Range<Token> tokenRange)
{
logger.debug("Calculating size estimate for {}.{} for range {}", keyspace, table, tokenRange);
try (Refs<SSTableReader> refs = RepairTokenRangeSplitter.getSSTableReaderRefs(repairType, keyspace, table, tokenRange))
{
SizeEstimate estimate = getSizesForRangeOfSSTables(repairType, keyspace, table, tokenRange, refs);
logger.debug("Generated size estimate {}", estimate);
return estimate;
}
}
/**
* Calculates the size estimation qualified to be repaired for a given {@code repairType},
* {@code keyspace}, {@code table}, {@code tokenRange}, and {@code refs}.
* <p>
* If the compression is enabled, then the size will be an estimate, otherwise it will be accurate.
* </p>
*
* @param repairType
* @param keyspace
* @param table
* @param tokenRange
* @param refs
* @return an estimate representing the number of partitions, size in range, and total size
*/
static SizeEstimate getSizesForRangeOfSSTables(RepairType repairType, String keyspace, String table,
Range<Token> tokenRange, Refs<SSTableReader> refs)
{
List<Range<Token>> singletonRange = Collections.singletonList(tokenRange);
ICardinality cardinality = new HyperLogLogPlus(13, 25);
long approxBytesInRange = 0L;
long totalBytes = 0L;
for (SSTableReader reader : refs)
{
try
{
if (reader.openReason == SSTableReader.OpenReason.EARLY)
continue;
CompactionMetadata metadata = (CompactionMetadata) reader.descriptor.getMetadataSerializer().deserialize(reader.descriptor, MetadataType.COMPACTION);
if (metadata != null)
cardinality = cardinality.merge(metadata.cardinalityEstimator);
// use onDiskLength, which is the actual size of the SSTable data file.
long sstableSize = reader.onDiskLength();
totalBytes += sstableSize;
// get the on disk size for the token range, note for compressed data this includes the full
// chunks the start and end ranges are found in.
long approximateRangeBytesInSSTable = reader.onDiskSizeForPartitionPositions(reader.getPositionsForRanges(singletonRange));
approxBytesInRange += Math.min(approximateRangeBytesInSSTable, sstableSize);
}
catch (IOException | CardinalityMergeException e)
{
logger.error("Error calculating size estimate for {}.{} for range {} on {}", keyspace, table, tokenRange, reader, e);
}
}
long partitions = 0L;
if (totalBytes > 0)
{
// use the ratio from size to estimate the partitions in the range as well
double ratio = approxBytesInRange / (double) totalBytes;
partitions = (long) Math.max(1, Math.ceil(cardinality.cardinality() * ratio));
}
return new SizeEstimate(repairType, keyspace, table, tokenRange, partitions, approxBytesInRange, totalBytes);
}
/**
* Calculates the token ranges owned by this node for a given keyspace.
*
* @param primaryRangeOnly whether to use only primary token ranges or include replicated ones
* @param keyspaceName the name of the keyspace
* @return one or more token ranges owned by this node
*/
static List<Range<Token>> getTokenRanges(boolean primaryRangeOnly, String keyspaceName)
{
// Collect all applicable token ranges
Collection<Range<Token>> wrappedRanges;
if (primaryRangeOnly)
{
wrappedRanges = TokenRingUtils.getPrimaryRangesForEndpoint(keyspaceName, FBUtilities.getBroadcastAddressAndPort());
}
else
{
wrappedRanges = StorageService.instance.getLocalRanges(keyspaceName);
}
// Unwrap each range as we need to account for ranges that overlap the ring
List<Range<Token>> ranges = new ArrayList<>();
for (Range<Token> wrappedRange : wrappedRanges)
{
ranges.addAll(wrappedRange.unwrap());
}
return ranges;
}
/**
* Calculates the total bytes to be repaired for a given keyspace and list of tables.
*
* @param repairType the repair type (e.g., FULL, INCREMENTAL)
* @param keyspaceName the name of the keyspace
* @param tableNames the list of tables
* @return a key-value map where the key is {@code keyspaceName.tableName} and the value is the number of bytes
* to be repaired.
*/
public static Map<String, Map<Range<Token>, SizeEstimate>> calcTotalBytesToBeRepaired(RepairType repairType, String keyspaceName, List<String> tableNames, List<Range<Token>> tokenRanges)
{
Map<String, Map<Range<Token>, SizeEstimate>> ksTablesEstimatedBytes = new HashMap<>();
for (String tableName : tableNames)
{
String ksTable = getKeyspaceTableName(keyspaceName, tableName);
ksTablesEstimatedBytes.computeIfAbsent(ksTable, k -> new HashMap<>());
Map<Range<Token>, SizeEstimate> tokenToSize = ksTablesEstimatedBytes.get(ksTable);
for (Range<Token> tokenRange : tokenRanges)
{
SizeEstimate tableAssignments = getRangeSizeEstimate(repairType, keyspaceName, tableName, tokenRange);
tokenToSize.put(tokenRange, tableAssignments);
}
}
return ksTablesEstimatedBytes;
}
public static String getKeyspaceTableName(String keyspace, String table)
{
return keyspace + "." + table;
}
/**
* Represents a size estimate by both bytes and partition count for a given keyspace and table for a token range.
*/
@VisibleForTesting
protected static class SizeEstimate
{
public final RepairType repairType;
public final String keyspace;
public final String table;
public final Range<Token> tokenRange;
public final long partitions;
public final long sizeInRange;
public final long totalSize;
/**
* Size to consider in the repair. For incremental repair, we want to consider the total size
* of the estimate as we have to factor in anticompacting the entire SSTable.
* For full repair, just use the size containing the range.
*/
public final long sizeForRepair;
public SizeEstimate(RepairType repairType,
String keyspace, String table, Range<Token> tokenRange,
long partitions, long sizeInRange, long totalSize)
{
this.repairType = repairType;
this.keyspace = keyspace;
this.table = table;
this.tokenRange = tokenRange;
this.partitions = partitions;
this.sizeInRange = sizeInRange;
this.totalSize = totalSize;
this.sizeForRepair = repairType == RepairType.INCREMENTAL ? totalSize : sizeInRange;
}
@Override
public String toString()
{
return "SizeEstimate{" +
"repairType=" + repairType +
", keyspace='" + keyspace + '\'' +
", table='" + table + '\'' +
", tokenRange=" + tokenRange +
", partitions=" + partitions +
", sizeInRange=" + sizeInRange +
", totalSize=" + totalSize +
", sizeForRepair=" + sizeForRepair +
'}';
}
}
}

23
src/java/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitter.java
View File

@ -32,10 +32,6 @@ import org.apache.cassandra.service.AutoRepairService;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.split;
/**
@ -98,13 +94,7 @@ public class FixedSplitTokenRangeSplitter implements IAutoRepairTokenRangeSplitt
String keyspaceName = repairPlan.getKeyspaceName();
List<String> tableNames = repairPlan.getTableNames();
Collection<Range<Token>> tokens = TokenRingUtils.getPrimaryRangesForEndpoint(keyspaceName, FBUtilities.getBroadcastAddressAndPort());
if (!primaryRangeOnly)
{
// if we need to repair non-primary token ranges, then change the tokens accordingly
tokens = StorageService.instance.getLocalReplicas(keyspaceName).onlyFull().ranges();
}
Collection<Range<Token>> tokens = AutoRepairUtils.getTokenRanges(primaryRangeOnly, keyspaceName);
boolean byKeyspace = config.getRepairByKeyspace(repairType);
// collect all token ranges.
List<Range<Token>> allRanges = new ArrayList<>();
@ -117,10 +107,15 @@ public class FixedSplitTokenRangeSplitter implements IAutoRepairTokenRangeSplitt
if (byKeyspace)
{
// This calculation is the best effort for the FixedSplitTokenRangeSplitter.
// In practice, this metric may not give you an accurate view in case of uneven data distribution.
long totalBytes = repairPlan.getEstimatedBytes();
long bytesPerRange = Math.max(1, totalBytes / splitsPerRange);
for (Range<Token> splitRange : allRanges)
{
// add repair assignment for each range entire keyspace's tables
repairAssignments.add(new RepairAssignment(splitRange, keyspaceName, tableNames));
repairAssignments.add(new RepairAssignment(splitRange, keyspaceName, tableNames, bytesPerRange));
}
}
else
@ -128,9 +123,11 @@ public class FixedSplitTokenRangeSplitter implements IAutoRepairTokenRangeSplitt
// add repair assignment per table
for (String tableName : tableNames)
{
long totalBytes = repairPlan.getTableEstimatedBytes(AutoRepairUtils.getKeyspaceTableName(keyspaceName, tableName));
long bytesPerRange = Math.max(1, totalBytes / splitsPerRange);
for (Range<Token> splitRange : allRanges)
{
repairAssignments.add(new RepairAssignment(splitRange, keyspaceName, Collections.singletonList(tableName)));
repairAssignments.add(new RepairAssignment(splitRange, keyspaceName, Collections.singletonList(tableName), bytesPerRange));
}
}
}

40
src/java/org/apache/cassandra/repair/autorepair/KeyspaceRepairPlan.java
View File

@ -18,9 +18,16 @@
package org.apache.cassandra.repair.autorepair;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
/**
* Encapsulates an intent to repair the given keyspace's tables
*/
@ -30,10 +37,14 @@ public class KeyspaceRepairPlan
private final List<String> tableNames;
public KeyspaceRepairPlan(String keyspaceName, List<String> tableNames)
@VisibleForTesting
public Map<String, Map<Range<Token>, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes;
public KeyspaceRepairPlan(String keyspaceName, List<String> tableNames, Map<String, Map<Range<Token>, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes)
{
this.keyspaceName = keyspaceName;
this.tableNames = tableNames;
this.ksTablesEstimatedBytes = ksTablesEstimatedBytes;
}
public String getKeyspaceName()
@ -46,18 +57,40 @@ public class KeyspaceRepairPlan
return tableNames;
}
public long getEstimatedBytes()
{
return ksTablesEstimatedBytes.values().stream()
.flatMap(tableMap -> tableMap.values().stream())
.mapToLong(sizeEstimate -> sizeEstimate.sizeForRepair)
.sum();
}
public long getTableEstimatedBytes(String keyspaceTableName)
{
return ksTablesEstimatedBytes.getOrDefault(keyspaceTableName,
Collections.emptyMap()).values().stream().mapToLong(sizeEstimate -> sizeEstimate.sizeForRepair).sum();
}
public AutoRepairUtils.SizeEstimate getSizeEstimate(String keyspaceTableName, Range<Token> tokenRange)
{
return ksTablesEstimatedBytes == null ? null
: ksTablesEstimatedBytes.getOrDefault(keyspaceTableName, null) == null ? null
: ksTablesEstimatedBytes.get(keyspaceTableName).get(tokenRange);
}
@Override
public boolean equals(Object o)
{
if (o == null || getClass() != o.getClass()) return false;
KeyspaceRepairPlan that = (KeyspaceRepairPlan) o;
return Objects.equals(keyspaceName, that.keyspaceName) && Objects.equals(tableNames, that.tableNames);
return Objects.equals(keyspaceName, that.keyspaceName) && Objects.equals(tableNames, that.tableNames)
&& Objects.equals(ksTablesEstimatedBytes, that.ksTablesEstimatedBytes);
}
@Override
public int hashCode()
{
return Objects.hash(keyspaceName, tableNames);
return Objects.hash(keyspaceName, tableNames, ksTablesEstimatedBytes);
}
@Override
@ -66,6 +99,7 @@ public class KeyspaceRepairPlan
return "KeyspaceRepairPlan{" +
"keyspaceName='" + keyspaceName + '\'' +
", tableNames=" + tableNames +
", ksTablesEstimatedBytes=" + ksTablesEstimatedBytes +
'}';
}
}

32
src/java/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlan.java
View File

@ -19,7 +19,6 @@
package org.apache.cassandra.repair.autorepair;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@ -89,11 +88,11 @@ public class PrioritizedRepairPlan
* for their given priority.
*
* @param keyspacesToTableNames A mapping keyspace to table names
* @param repairType The repair type that is being executed
* @param orderFunc A function to order keyspace and tables in the returned plan.
* @param repairType The repair type that is being executed
* @param orderFunc A function to order keyspace and tables in the returned plan.
* @return Ordered list of plan's by table priorities.
*/
public static List<PrioritizedRepairPlan> build(Map<String, List<String>> keyspacesToTableNames, AutoRepairConfig.RepairType repairType, Consumer<List<String>> orderFunc)
public static List<PrioritizedRepairPlan> build(Map<String, List<String>> keyspacesToTableNames, AutoRepairConfig.RepairType repairType, Consumer<List<String>> orderFunc, boolean primaryRangeOnly)
{
// Build a map of priority -> (keyspace -> tables)
Map<Integer, Map<String, List<String>>> plans = new HashMap<>();
@ -124,31 +123,20 @@ public class PrioritizedRepairPlan
List<String> keyspaceNames = new ArrayList<>(keyspacesAndTables.keySet());
orderFunc.accept(keyspaceNames);
for(String keyspaceName : keyspaceNames)
for (String keyspaceName : keyspaceNames)
{
List<String> tableNames = keyspacesAndTables.get(keyspaceName);
orderFunc.accept(tableNames);
KeyspaceRepairPlan keyspaceRepairPlan = new KeyspaceRepairPlan(keyspaceName, new ArrayList<>(tableNames));
keyspaceRepairPlans.add(keyspaceRepairPlan);
List<String> tableNames = keyspacesAndTables.get(keyspaceName);
orderFunc.accept(tableNames);
KeyspaceRepairPlan keyspaceRepairPlan =
new KeyspaceRepairPlan(keyspaceName, new ArrayList<>(tableNames),
AutoRepairUtils.calcTotalBytesToBeRepaired(repairType, keyspaceName, tableNames, AutoRepairUtils.getTokenRanges(primaryRangeOnly, keyspaceName)));
keyspaceRepairPlans.add(keyspaceRepairPlan);
}
}
return planList;
}
/**
* Convenience method to build a repair plan for a single keyspace with tables. Primarily useful in testing.
* @param keyspaceName Keyspace to repair
* @param tableNames tables to repair for the given keyspace.
* @return Single repair plan.
*/
static List<PrioritizedRepairPlan> buildSingleKeyspacePlan(AutoRepairConfig.RepairType repairType, String keyspaceName, String ... tableNames)
{
Map<String, List<String>> keyspaceMap = new HashMap<>();
keyspaceMap.put(keyspaceName, Arrays.asList(tableNames));
return build(keyspaceMap, repairType, (l) -> {});
}
/**
* @return The priority of the given table if defined, otherwise 0.
*/

16
src/java/org/apache/cassandra/repair/autorepair/RepairAssignment.java
View File

@ -35,11 +35,14 @@ public class RepairAssignment
final List<String> tableNames;
public RepairAssignment(Range<Token> tokenRange, String keyspaceName, List<String> tableNames)
protected final long estimatedBytes;
public RepairAssignment(Range<Token> tokenRange, String keyspaceName, List<String> tableNames, long estimatedBytes)
{
this.tokenRange = tokenRange;
this.keyspaceName = keyspaceName;
this.tableNames = tableNames;
this.estimatedBytes = estimatedBytes;
}
public Range<Token> getTokenRange()
@ -57,19 +60,25 @@ public class RepairAssignment
return tableNames;
}
public long getEstimatedBytes()
{
return estimatedBytes;
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
RepairAssignment that = (RepairAssignment) o;
return Objects.equals(tokenRange, that.tokenRange) && Objects.equals(keyspaceName, that.keyspaceName) && Objects.equals(tableNames, that.tableNames);
return Objects.equals(tokenRange, that.tokenRange) && Objects.equals(keyspaceName, that.keyspaceName)
&& Objects.equals(tableNames, that.tableNames) && Objects.equals(estimatedBytes, that.estimatedBytes);
}
@Override
public int hashCode()
{
return Objects.hash(tokenRange, keyspaceName, tableNames);
return Objects.hash(tokenRange, keyspaceName, tableNames, estimatedBytes);
}
@Override
@ -79,6 +88,7 @@ public class RepairAssignment
"tokenRange=" + tokenRange +
", keyspaceName='" + keyspaceName + '\'' +
", tableNames=" + tableNames +
", estimatedBytes=" + estimatedBytes +
'}';
}
}

283
src/java/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitter.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.repair.autorepair;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@ -39,14 +38,9 @@ import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.clearspring.analytics.stream.cardinality.CardinalityMergeException;
import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
import com.clearspring.analytics.stream.cardinality.ICardinality;
import org.apache.cassandra.config.DataStorageSpec;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.PartitionPosition;
@ -56,11 +50,8 @@ import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.metadata.CompactionMetadata;
import org.apache.cassandra.io.sstable.metadata.MetadataType;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.service.AutoRepairService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.split;
@ -228,7 +219,8 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
* A custom {@link RepairAssignmentIterator} that confines the number of repair assignments to
* <code>max_bytes_per_schedule</code>.
*/
private class BytesBasedRepairAssignmentIterator extends RepairAssignmentIterator {
private class BytesBasedRepairAssignmentIterator extends RepairAssignmentIterator
{
private final boolean primaryRangeOnly;
private long bytesSoFar = 0;
@ -250,14 +242,14 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
return new KeyspaceRepairAssignments(priority, repairPlan.getKeyspaceName(), Collections.emptyList());
}
List<Range<Token>> tokenRanges = getTokenRanges(primaryRangeOnly, repairPlan.getKeyspaceName());
List<Range<Token>> tokenRanges = AutoRepairUtils.getTokenRanges(primaryRangeOnly, repairPlan.getKeyspaceName());
// shuffle token ranges to unbias selection of ranges
Collections.shuffle(tokenRanges);
List<SizedRepairAssignment> repairAssignments = new ArrayList<>();
// Generate assignments for each range speparately
for (Range<Token> tokenRange : tokenRanges)
{
repairAssignments.addAll(getRepairAssignmentsForKeyspace(repairType, repairPlan.getKeyspaceName(), repairPlan.getTableNames(), tokenRange));
repairAssignments.addAll(getRepairAssignmentsForKeyspace(repairType, repairPlan, tokenRange));
}
FilteredRepairAssignments filteredRepairAssignments = filterRepairAssignments(priority, repairPlan.getKeyspaceName(), repairAssignments, bytesSoFar);
@ -267,7 +259,7 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
}
@VisibleForTesting
List<SizedRepairAssignment> getRepairAssignmentsForKeyspace(AutoRepairConfig.RepairType repairType, String keyspaceName, List<String> tableNames, Range<Token> tokenRange)
List<SizedRepairAssignment> getRepairAssignmentsForKeyspace(AutoRepairConfig.RepairType repairType, KeyspaceRepairPlan repairPlan, Range<Token> tokenRange)
{
List<SizedRepairAssignment> repairAssignments = new ArrayList<>();
// this is used for batching minimal single assignment tables together
@ -277,12 +269,12 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
// If we can repair by keyspace, sort the tables by size so can batch the smallest ones together
boolean repairByKeyspace = config.getRepairByKeyspace(repairType);
List<String> tablesToProcess = tableNames;
List<String> tablesToProcess = repairPlan.getTableNames();
if (repairByKeyspace)
{
tablesToProcess = tableNames.stream().sorted((t1, t2) -> {
ColumnFamilyStore cfs1 = ColumnFamilyStore.getIfExists(keyspaceName, t1);
ColumnFamilyStore cfs2 = ColumnFamilyStore.getIfExists(keyspaceName, t2);
tablesToProcess = repairPlan.getTableNames().stream().sorted((t1, t2) -> {
ColumnFamilyStore cfs1 = ColumnFamilyStore.getIfExists(repairPlan.getKeyspaceName(), t1);
ColumnFamilyStore cfs2 = ColumnFamilyStore.getIfExists(repairPlan.getKeyspaceName(), t2);
// If for whatever reason the CFS is not retrievable, we can assume it has been deleted, so give the
// other cfs precedence.
if (cfs1 == null)
@ -301,7 +293,7 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
for (String tableName : tablesToProcess)
{
List<SizedRepairAssignment> tableAssignments = getRepairAssignmentsForTable(keyspaceName, tableName, tokenRange);
List<SizedRepairAssignment> tableAssignments = getRepairAssignmentsForTable(repairPlan, tableName, tokenRange);
if (tableAssignments.isEmpty())
continue;
@ -348,8 +340,9 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
/**
* Given a repair type and map of sized-based repair assignments, confine them by <code>maxBytesPerSchedule</code>.
*
* @param repairAssignments the assignments to filter.
* @param bytesSoFar repair assignment bytes accumulated so far.
* @param bytesSoFar repair assignment bytes accumulated so far.
* @return A list of repair assignments confined by <code>maxBytesPerSchedule</code>.
*/
@VisibleForTesting
@ -444,9 +437,9 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
}
/**
* @param repairAssignments The assignments to sum
* @return The sum of {@link SizedRepairAssignment#getEstimatedBytes()} of all given
* repairAssignments.
* @param repairAssignments The assignments to sum
*/
@VisibleForTesting
protected static long getEstimatedBytes(List<SizedRepairAssignment> repairAssignments)
@ -484,10 +477,18 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
}
@VisibleForTesting
protected List<SizedRepairAssignment> getRepairAssignmentsForTable(String keyspaceName, String tableName, Range<Token> tokenRange)
protected List<SizedRepairAssignment> getRepairAssignmentsForTable(KeyspaceRepairPlan repairPlan, String tableName, Range<Token> tokenRange)
{
List<SizeEstimate> sizeEstimates = getRangeSizeEstimate(keyspaceName, tableName, tokenRange);
return getRepairAssignments(sizeEstimates);
AutoRepairUtils.SizeEstimate sizeEstimate = repairPlan.getSizeEstimate(AutoRepairUtils.getKeyspaceTableName(repairPlan.getKeyspaceName(), tableName), tokenRange);
if (sizeEstimate == null)
{
// Ideally, it should have been cached already inside the KeyspaceRepairPlan, but incase it was not,
// then recalculating it. It is a bit expensive, but necessary for the repair
logger.warn("The size estimate for {}.{} range {} was not pre-calculated, calculating on-demand",
repairPlan.getKeyspaceName(), tableName, tokenRange);
sizeEstimate = AutoRepairUtils.getRangeSizeEstimate(repairType, repairPlan.getKeyspaceName(), tableName, tokenRange);
}
return getRepairAssignments(sizeEstimate);
}
private static void logSkippingTable(String keyspaceName, String tableName)
@ -496,82 +497,75 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
}
@VisibleForTesting
protected List<SizedRepairAssignment> getRepairAssignments(List<SizeEstimate> sizeEstimates)
protected List<SizedRepairAssignment> getRepairAssignments(AutoRepairUtils.SizeEstimate estimate)
{
List<SizedRepairAssignment> repairAssignments = new ArrayList<>();
// since its possible for us to hit maxBytesPerSchedule before seeing all ranges, shuffle so there is chance
// at least of hitting all the ranges _eventually_ for the worst case scenarios
Collections.shuffle(sizeEstimates);
int totalExpectedSubRanges = 0;
for (SizeEstimate estimate : sizeEstimates)
if (estimate.sizeForRepair != 0)
{
if (estimate.sizeForRepair != 0)
boolean needsSplitting = estimate.sizeForRepair > bytesPerAssignment.toBytes() || estimate.partitions > partitionsPerAssignment;
if (needsSplitting)
{
boolean needsSplitting = estimate.sizeForRepair > bytesPerAssignment.toBytes() || estimate.partitions > partitionsPerAssignment;
if (needsSplitting)
{
totalExpectedSubRanges += calculateNumberOfSplits(estimate);
}
totalExpectedSubRanges += calculateNumberOfSplits(estimate);
}
}
for (SizeEstimate estimate : sizeEstimates)
if (estimate.sizeForRepair == 0)
{
if (estimate.sizeForRepair == 0)
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(estimate.keyspace, estimate.table);
if (cfs == null)
{
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(estimate.keyspace, estimate.table);
logSkippingTable(estimate.keyspace, estimate.table);
return Collections.emptyList();
}
if (cfs == null)
long memtableSize = cfs.getTracker().getView().getCurrentMemtable().getLiveDataSize();
if (memtableSize > 0L)
{
logger.debug("Included {}.{} range {}, had no unrepaired SSTables, but memtableSize={}, adding single repair assignment", estimate.keyspace, estimate.table, estimate.tokenRange, memtableSize);
SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, Collections.singletonList(estimate.table), "full primary rangee for table with memtable only detected", memtableSize);
repairAssignments.add(assignment);
}
else
{
logger.debug("Included {}.{} range {}, has no SSTables or memtable data, but adding single repair assignment for entire range in case writes were missed", estimate.keyspace, estimate.table, estimate.tokenRange);
SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, Collections.singletonList(estimate.table), "full primary range for table with no data detected", 0L);
repairAssignments.add(assignment);
}
}
else
{
// Check if the estimate needs splitting based on the criteria
boolean needsSplitting = estimate.sizeForRepair > bytesPerAssignment.toBytes() || estimate.partitions > partitionsPerAssignment;
if (needsSplitting)
{
int numberOfSplits = calculateNumberOfSplits(estimate);
long approximateBytesPerSplit = estimate.sizeForRepair / numberOfSplits;
Collection<Range<Token>> subranges = split(estimate.tokenRange, numberOfSplits);
for (Range<Token> subrange : subranges)
{
logSkippingTable(estimate.keyspace, estimate.table);
continue;
}
long memtableSize = cfs.getTracker().getView().getCurrentMemtable().getLiveDataSize();
if (memtableSize > 0L)
{
logger.debug("Included {}.{} range {}, had no unrepaired SSTables, but memtableSize={}, adding single repair assignment", estimate.keyspace, estimate.table, estimate.tokenRange, memtableSize);
SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, Collections.singletonList(estimate.table), "full primary rangee for table with memtable only detected", memtableSize);
repairAssignments.add(assignment);
}
else
{
logger.debug("Included {}.{} range {}, has no SSTables or memtable data, but adding single repair assignment for entire range in case writes were missed", estimate.keyspace, estimate.table, estimate.tokenRange);
SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace, Collections.singletonList(estimate.table), "full primary range for table with no data detected", 0L);
SizedRepairAssignment assignment = new SizedRepairAssignment(subrange, estimate.keyspace, Collections.singletonList(estimate.table),
String.format("subrange %d of %d", repairAssignments.size() + 1, totalExpectedSubRanges),
approximateBytesPerSplit);
repairAssignments.add(assignment);
}
}
else
{
// Check if the estimate needs splitting based on the criteria
boolean needsSplitting = estimate.sizeForRepair > bytesPerAssignment.toBytes() || estimate.partitions > partitionsPerAssignment;
if (needsSplitting)
{
int numberOfSplits = calculateNumberOfSplits(estimate);
long approximateBytesPerSplit = estimate.sizeForRepair / numberOfSplits;
Collection<Range<Token>> subranges = split(estimate.tokenRange, numberOfSplits);
for (Range<Token> subrange : subranges)
{
SizedRepairAssignment assignment = new SizedRepairAssignment(subrange, estimate.keyspace, Collections.singletonList(estimate.table),
String.format("subrange %d of %d", repairAssignments.size()+1, totalExpectedSubRanges),
approximateBytesPerSplit);
repairAssignments.add(assignment);
}
}
else
{
// No splitting needed, repair the entire range as-is
SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace,
Collections.singletonList(estimate.table),
"full primary range for table", estimate.sizeForRepair);
repairAssignments.add(assignment);
}
// No splitting needed, repair the entire range as-is
SizedRepairAssignment assignment = new SizedRepairAssignment(estimate.tokenRange, estimate.keyspace,
Collections.singletonList(estimate.table),
"full primary range for table", estimate.sizeForRepair);
repairAssignments.add(assignment);
}
}
return repairAssignments;
}
private int calculateNumberOfSplits(SizeEstimate estimate)
private int calculateNumberOfSplits(AutoRepairUtils.SizeEstimate estimate)
{
// Calculate the number of splits needed for size and partitions
int splitsForSize = (int) Math.ceil((double) estimate.sizeForRepair / bytesPerAssignment.toBytes());
@ -597,84 +591,6 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
return splits;
}
private List<Range<Token>> getTokenRanges(boolean primaryRangeOnly, String keyspaceName)
{
// Collect all applicable token ranges
Collection<Range<Token>> wrappedRanges;
if (primaryRangeOnly)
{
wrappedRanges = TokenRingUtils.getPrimaryRangesForEndpoint(keyspaceName, FBUtilities.getBroadcastAddressAndPort());
}
else
{
wrappedRanges = StorageService.instance.getLocalRanges(keyspaceName);
}
// Unwrap each range as we need to account for ranges that overlap the ring
List<Range<Token>> ranges = new ArrayList<>();
for (Range<Token> wrappedRange : wrappedRanges)
{
ranges.addAll(wrappedRange.unwrap());
}
return ranges;
}
private List<SizeEstimate> getRangeSizeEstimate(String keyspace, String table, Range<Token> tokenRange)
{
List<SizeEstimate> sizeEstimates = new ArrayList<>();
logger.debug("Calculating size estimate for {}.{} for range {}", keyspace, table, tokenRange);
try (Refs<SSTableReader> refs = getSSTableReaderRefs(repairType, keyspace, table, tokenRange))
{
SizeEstimate estimate = getSizesForRangeOfSSTables(repairType, keyspace, table, tokenRange, refs);
logger.debug("Generated size estimate {}", estimate);
sizeEstimates.add(estimate);
}
return sizeEstimates;
}
@VisibleForTesting
static SizeEstimate getSizesForRangeOfSSTables(AutoRepairConfig.RepairType repairType, String keyspace, String table, Range<Token> tokenRange, Refs<SSTableReader> refs)
{
List<Range<Token>> singletonRange = Collections.singletonList(tokenRange);
ICardinality cardinality = new HyperLogLogPlus(13, 25);
long approxBytesInRange = 0L;
long totalBytes = 0L;
for (SSTableReader reader : refs)
{
try
{
if (reader.openReason == SSTableReader.OpenReason.EARLY)
continue;
CompactionMetadata metadata = (CompactionMetadata) reader.descriptor.getMetadataSerializer().deserialize(reader.descriptor, MetadataType.COMPACTION);
if (metadata != null)
cardinality = cardinality.merge(metadata.cardinalityEstimator);
// use onDiskLength, which is the actual size of the SSTable data file.
long sstableSize = reader.onDiskLength();
totalBytes += sstableSize;
// get the on disk size for the token range, note for compressed data this includes the full
// chunks the start and end ranges are found in.
long approximateRangeBytesInSSTable = reader.onDiskSizeForPartitionPositions(reader.getPositionsForRanges(singletonRange));
approxBytesInRange += Math.min(approximateRangeBytesInSSTable, sstableSize);
}
catch (IOException | CardinalityMergeException e)
{
logger.error("Error calculating size estimate for {}.{} for range {} on {}", keyspace, table, tokenRange, reader, e);
}
}
long partitions = 0L;
if (totalBytes > 0)
{
// use the ratio from size to estimate the partitions in the range as well
double ratio = approxBytesInRange / (double) totalBytes;
partitions = (long) Math.max(1, Math.ceil(cardinality.cardinality() * ratio));
}
return new SizeEstimate(repairType, keyspace, table, tokenRange, partitions, approxBytesInRange, totalBytes);
}
@VisibleForTesting
static Refs<SSTableReader> getSSTableReaderRefs(AutoRepairConfig.RepairType repairType, String keyspaceName, String tableName, Range<Token> tokenRange)
{
@ -749,66 +665,15 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
return Collections.unmodifiableMap(parameters);
}
/**
* Represents a size estimate by both bytes and partition count for a given keyspace and table for a token range.
*/
@VisibleForTesting
protected static class SizeEstimate
{
public final AutoRepairConfig.RepairType repairType;
public final String keyspace;
public final String table;
public final Range<Token> tokenRange;
public final long partitions;
public final long sizeInRange;
public final long totalSize;
/**
* Size to consider in the repair. For incremental repair, we want to consider the total size
* of the estimate as we have to factor in anticompacting the entire SSTable.
* For full repair, just use the size containing the range.
*/
public final long sizeForRepair;
public SizeEstimate(AutoRepairConfig.RepairType repairType,
String keyspace, String table, Range<Token> tokenRange,
long partitions, long sizeInRange, long totalSize)
{
this.repairType = repairType;
this.keyspace = keyspace;
this.table = table;
this.tokenRange = tokenRange;
this.partitions = partitions;
this.sizeInRange = sizeInRange;
this.totalSize = totalSize;
this.sizeForRepair = repairType == AutoRepairConfig.RepairType.INCREMENTAL ? totalSize : sizeInRange;
}
@Override
public String toString()
{
return "SizeEstimate{" +
"repairType=" + repairType +
", keyspace='" + keyspace + '\'' +
", table='" + table + '\'' +
", tokenRange=" + tokenRange +
", partitions=" + partitions +
", sizeInRange=" + sizeInRange +
", totalSize=" + totalSize +
", sizeForRepair=" + sizeForRepair +
'}';
}
}
/**
* Implementation of RepairAssignment that also assigns an estimation of bytes involved
* in the repair.
*/
@VisibleForTesting
protected static class SizedRepairAssignment extends RepairAssignment {
protected static class SizedRepairAssignment extends RepairAssignment
{
final String description;
final long estimatedBytes;
public SizedRepairAssignment(Range<Token> tokenRange, String keyspaceName, List<String> tableNames)
{
@ -819,9 +684,8 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
String description,
long estimatedBytes)
{
super(tokenRange, keyspaceName, tableNames);
super(tokenRange, keyspaceName, tableNames, estimatedBytes);
this.description = description;
this.estimatedBytes = estimatedBytes;
}
/**
@ -833,7 +697,8 @@ public class RepairTokenRangeSplitter implements IAutoRepairTokenRangeSplitter
}
/**
* Estimated bytes involved in the assignment. Typically Derived from {@link SizeEstimate#sizeForRepair}.
* Estimated bytes involved in the assignment. Typically Derived from {@link AutoRepairUtils.SizeEstimate#sizeForRepair}.
*
* @return estimated bytes involved in the assignment.
*/
public long getEstimatedBytes()

234
test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsHelper.java Normal file
View File

@ -0,0 +1,234 @@
/*
* 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.repair;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableMap;
import org.junit.Assert;
import org.apache.cassandra.Util;
import org.apache.cassandra.auth.AuthKeyspace;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.metrics.AutoRepairMetrics;
import org.apache.cassandra.metrics.AutoRepairMetricsManager;
import org.apache.cassandra.repair.autorepair.AutoRepair;
import org.apache.cassandra.repair.autorepair.AutoRepairConfig;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.SystemDistributedKeyspace;
import org.apache.cassandra.service.AutoRepairService;
import static org.apache.cassandra.schema.SchemaConstants.DISTRIBUTED_KEYSPACE_NAME;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
/**
* Helper class to test {@code totalBytesToRepair}, {@code bytesAlreadyRepaired}, {@code totalKeyspaceRepairPlansToRepair},
* and {@code keyspaceRepairPlansAlreadyRepaired}
* for {@link org.apache.cassandra.repair.autorepair.AutoRepairState} scheduler
*/
public class AutoRepairSchedulerStatsHelper extends TestBaseImpl
{
private static Cluster cluster;
static SimpleDateFormat sdf;
private static final String KEYSPACE1 = "ks1";
private static final String KEYSPACE2 = "ks2";
private static final String TABLE1 = "tbl1";
private static final String TABLE2 = "tbl2";
public static void init(int numTokens) throws IOException
{
// Define the expected date format pattern
String pattern = "EEE MMM dd HH:mm:ss z yyyy";
// Create SimpleDateFormat object with the given pattern
sdf = new SimpleDateFormat(pattern);
sdf.setLenient(false);
CassandraRelevantProperties.SYSTEM_DISTRIBUTED_DEFAULT_RF.setInt(1);
cluster = Cluster.build(1)
.withTokenCount(numTokens)
.withTokenSupplier(TokenSupplier.evenlyDistributedTokens(1, numTokens))
.withConfig(config -> config
.set("num_tokens", numTokens)
.set("auto_repair",
ImmutableMap.of(
"repair_type_overrides",
ImmutableMap.of(AutoRepairConfig.RepairType.FULL.getConfigName(),
ImmutableMap.of(
"initial_scheduler_delay", "5s",
"enabled", "true",
"parallel_repair_count", "1",
// Allow parallel replica repair to allow replicas
// to execute full repair at same time.
"allow_parallel_replica_repair", "true",
// Set min_repair_interval to a higher number to
// run only one round of AutoRepair
"min_repair_interval", "48h"))))
.set("auto_repair.enabled", "true")
.set("auto_repair.global_settings.repair_retry_backoff", "5s")
.set("auto_repair.repair_task_min_duration", "0s")
.set("auto_repair.repair_check_interval", "5s"))
.start();
cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE1 + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE2 + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
// disable the compression to calculate an accurate expected repair bytes because with compression enabled,
// we only get estimated bytes, which hinders the ability to do actual vs. expected checks in the test case
cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compression = { 'enabled' : false }", KEYSPACE1, TABLE1));
cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compression = { 'enabled' : false }", KEYSPACE1, TABLE2));
cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compression = { 'enabled' : false }", KEYSPACE2, TABLE1));
cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compression = { 'enabled' : false }", KEYSPACE2, TABLE2));
}
public static void tearDown()
{
cluster.close();
}
public static void testSchedulerStats() throws ParseException
{
// ensure there was no history of previous repair runs through the scheduler
Object[][] rows = cluster.coordinator(1).execute(String.format("SELECT repair_type, host_id, repair_start_ts, repair_finish_ts, repair_turn FROM %s.%s", DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY), ConsistencyLevel.QUORUM);
assertEquals(0, rows.length);
// disabling AutoRepair for system_distributed and system_auth tables to avoid
// interfering with the repaired bytes/plans calculation
disableAutoRepair(SystemDistributedKeyspace.NAME, SystemDistributedKeyspace.TABLE_NAMES);
disableAutoRepair(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.TABLE_NAMES);
insertData();
cluster.get(1).runOnInstance(() -> {
try
{
AutoRepairService.setup();
AutoRepair.instance.setup();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
});
cluster.forEach(i -> i.runOnInstance(() -> {
AutoRepair.SLEEP_IF_REPAIR_FINISHES_QUICKLY = new DurationSpec.IntSecondsBound("2s");
AutoRepairMetrics fullMetrics = AutoRepairMetricsManager.getMetrics(AutoRepairConfig.RepairType.FULL);
// Since the AutoRepair sleeps up to SLEEP_IF_REPAIR_FINISHES_QUICKLY if the repair finishes quickly,
// so the "nodeRepairTimeInSec" metric should at least be greater than or equal to
// SLEEP_IF_REPAIR_FINISHES_QUICKLY
Util.spinAssert("AutoRepair has not yet completed one FULL repair cycle",
greaterThanOrEqualTo(2L),
() -> fullMetrics.nodeRepairTimeInSec.getValue().longValue(),
2,
TimeUnit.MINUTES);
long expectedRepairBytes = calculateExpectedBytes(Arrays.asList(KEYSPACE1, KEYSPACE2));
assertEquals(fullMetrics.totalKeyspaceRepairPlansToRepair.getValue(), fullMetrics.keyspaceRepairPlansAlreadyRepaired.getValue());
// AutoRepair creates a repair plan per keyspace;
// Since there are two separate keyspaces, KEYSPACE1 and KEYSPACE2, the total expected plans should be "2"
assertEquals(2, fullMetrics.totalKeyspaceRepairPlansToRepair.getValue().intValue());
assertEquals(fullMetrics.totalBytesToRepair.getValue().longValue(), fullMetrics.bytesAlreadyRepaired.getValue().longValue());
assertEquals(expectedRepairBytes, fullMetrics.bytesAlreadyRepaired.getValue().longValue());
}));
validate(AutoRepairConfig.RepairType.FULL.toString());
}
private static long calculateExpectedBytes(List<String> keyspaces)
{
long totalBytes = 0;
for (String keyspace : keyspaces)
{
for (String table : Arrays.asList(TABLE1, TABLE2))
{
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(keyspace, table);
assertNotNull(cfs);
Iterable<SSTableReader> sstables = cfs.getTracker().getView().select(SSTableSet.CANONICAL);
for (SSTableReader sstable : sstables)
{
totalBytes += sstable.onDiskLength();
}
}
}
return totalBytes;
}
private static void insertData()
{
for (int i = 0; i < 100; i++)
{
cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (?,?,?)", KEYSPACE1, TABLE1),
ConsistencyLevel.ONE, i, i, i);
cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (?,?,?)", KEYSPACE1, TABLE2),
ConsistencyLevel.ONE, i, i, i);
cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (?,?,?)", KEYSPACE2, TABLE1),
ConsistencyLevel.ONE, i, i, i);
cluster.coordinator(1).execute(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (?,?,?)", KEYSPACE2, TABLE2),
ConsistencyLevel.ONE, i, i, i);
}
cluster.get(1).nodetool("flush", KEYSPACE1, TABLE1);
cluster.get(1).nodetool("flush", KEYSPACE1, TABLE2);
cluster.get(1).nodetool("flush", KEYSPACE2, TABLE1);
cluster.get(1).nodetool("flush", KEYSPACE2, TABLE2);
}
private static void disableAutoRepair(String keyspaceName, Set<String> distributedSystemTables)
{
for (String tableName : distributedSystemTables)
{
cluster.coordinator(1).execute(String.format("ALTER TABLE %s.%s WITH auto_repair = {'full_enabled': 'false'}", keyspaceName, tableName),
ConsistencyLevel.ONE);
}
}
private static void validate(String repairType) throws ParseException
{
Object[][] rows = cluster.coordinator(1).execute(String.format("SELECT repair_type, host_id, repair_start_ts, repair_finish_ts, repair_turn FROM %s.%s where repair_type='%s'", DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, repairType), ConsistencyLevel.QUORUM);
assertEquals(1, rows.length);
for (int node = 0; node < rows.length; node++)
{
Object[] row = rows[node];
// repair_type
Assert.assertEquals(repairType, row[0].toString());
// host_id
Assert.assertNotNull(UUID.fromString(row[1].toString()));
// ensure there is a legit repair_start_ts and repair_finish_ts
sdf.parse(row[2].toString());
sdf.parse(row[3].toString());
// the reason why the repair was scheduled
Assert.assertNotNull(row[4]);
Assert.assertEquals("MY_TURN", row[4].toString());
}
}
}

54
test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsNoVNodesTest.java Normal file
View File

@ -0,0 +1,54 @@
/*
* 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.repair;
import java.io.IOException;
import java.text.ParseException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.distributed.test.TestBaseImpl;
/**
* Test {@code totalBytesToRepair}, {@code bytesAlreadyRepaired}, {@code totalKeyspaceRepairPlansToRepair},
* and {@code keyspaceRepairPlansAlreadyRepaired}
* for {@link org.apache.cassandra.repair.autorepair.AutoRepairState} scheduler without v-nodes
*/
public class AutoRepairSchedulerStatsNoVNodesTest extends TestBaseImpl
{
@BeforeClass
public static void init() throws IOException
{
AutoRepairSchedulerStatsHelper.init(1);
}
@AfterClass
public static void tearDown()
{
AutoRepairSchedulerStatsHelper.tearDown();
}
@Test
public void testSchedulerStats() throws ParseException
{
AutoRepairSchedulerStatsHelper.testSchedulerStats();
}
}

54
test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsVNodesTest.java Normal file
View File

@ -0,0 +1,54 @@
/*
* 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.repair;
import java.io.IOException;
import java.text.ParseException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.distributed.test.TestBaseImpl;
/**
* Test {@code totalBytesToRepair}, {@code bytesAlreadyRepaired}, {@code totalKeyspaceRepairPlansToRepair},
* and {@code keyspaceRepairPlansAlreadyRepaired}
* for {@link org.apache.cassandra.repair.autorepair.AutoRepairState} scheduler with v-nodes
*/
public class AutoRepairSchedulerStatsVNodesTest extends TestBaseImpl
{
@BeforeClass
public static void init() throws IOException
{
AutoRepairSchedulerStatsHelper.init(16);
}
@AfterClass
public static void tearDown()
{
AutoRepairSchedulerStatsHelper.tearDown();
}
@Test
public void testSchedulerStats() throws ParseException
{
AutoRepairSchedulerStatsHelper.testSchedulerStats();
}
}

8
test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateFactoryTest.java
View File

@ -34,15 +34,15 @@ public class AutoRepairStateFactoryTest
@Test
public void testGetRepairState()
{
AutoRepairState state = RepairType.getAutoRepairState(RepairType.FULL);
AutoRepairState state = RepairType.getAutoRepairState(RepairType.FULL, new AutoRepairConfig());
assertTrue(state instanceof FullRepairState);
state = RepairType.getAutoRepairState(RepairType.INCREMENTAL);
state = RepairType.getAutoRepairState(RepairType.INCREMENTAL, new AutoRepairConfig());
assertTrue(state instanceof IncrementalRepairState);
state = RepairType.getAutoRepairState(RepairType.PREVIEW_REPAIRED);
state = RepairType.getAutoRepairState(RepairType.PREVIEW_REPAIRED, new AutoRepairConfig());
assertTrue(state instanceof PreviewRepairedState);
}
@ -54,7 +54,7 @@ public class AutoRepairStateFactoryTest
{
try
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
assertNotNull(state);
} catch (IllegalArgumentException e)
{

53
test/unit/org/apache/cassandra/repair/autorepair/AutoRepairStateTest.java
View File

@ -34,8 +34,6 @@ import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.repair.autorepair.AutoRepairConfig.RepairType;
import org.apache.cassandra.repair.autorepair.AutoRepairUtils.AutoRepairHistory;
import org.apache.cassandra.service.AutoRepairService;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.mockito.Mock;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@ -54,9 +52,6 @@ public class AutoRepairStateTest extends CQLTester
@Parameterized.Parameter
public RepairType repairType;
@Mock
ProgressEvent progressEvent;
@Parameterized.Parameters
public static Collection<RepairType> repairTypes()
{
@ -74,7 +69,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetRepairRunnable()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
AutoRepairService.setup();
Runnable runnable = state.getRepairRunnable(KEYSPACE, ImmutableList.of(testTable), ImmutableSet.of(), false);
@ -85,7 +80,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetLastRepairTime()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.lastRepairTimeInMs = 1;
assertEquals(1, state.getLastRepairTime());
@ -94,7 +89,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetTotalTablesConsideredForRepair()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setTotalTablesConsideredForRepair(1);
@ -104,7 +99,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetTotalTablesConsideredForRepair()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.totalTablesConsideredForRepair = 1;
assertEquals(1, state.getTotalTablesConsideredForRepair());
@ -113,7 +108,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetLastRepairTimeInMs()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setLastRepairTime(1);
@ -123,7 +118,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetClusterRepairTimeInSec()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.clusterRepairTimeInSec = 1;
assertEquals(1, state.getClusterRepairTimeInSec());
@ -132,7 +127,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetNodeRepairTimeInSec()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.nodeRepairTimeInSec = 1;
assertEquals(1, state.getNodeRepairTimeInSec());
@ -141,7 +136,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetRepairInProgress()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setRepairInProgress(true);
@ -151,7 +146,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testIsRepairInProgress()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.repairInProgress = true;
assertTrue(state.isRepairInProgress());
@ -160,7 +155,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetSkippedTokenRangesCount()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setSkippedTokenRangesCount(1);
@ -170,7 +165,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetSkippedTokenRangesCount()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.skippedTokenRangesCount = 1;
assertEquals(1, state.getSkippedTokenRangesCount());
@ -179,7 +174,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetLongestUnrepairedSecNull()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.longestUnrepairedNode = null;
try
@ -195,7 +190,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetLongestUnrepairedSec()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.longestUnrepairedNode = new AutoRepairHistory(UUID.randomUUID(), "", 0, 1000,
null, 0, false);
AutoRepairState.timeFunc = () -> 2000L;
@ -213,7 +208,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetTotalMVTablesConsideredForRepair()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setTotalMVTablesConsideredForRepair(1);
@ -223,7 +218,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetTotalMVTablesConsideredForRepair()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.totalMVTablesConsideredForRepair = 1;
assertEquals(1, state.getTotalMVTablesConsideredForRepair());
@ -232,7 +227,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetNodeRepairTimeInSec()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setNodeRepairTimeInSec(1);
@ -242,7 +237,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetClusterRepairTimeInSec()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setClusterRepairTimeInSec(1);
@ -252,7 +247,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetRepairKeyspaceCount()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setRepairKeyspaceCount(1);
@ -262,7 +257,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetRepairKeyspaceCount()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.repairKeyspaceCount = 1;
assertEquals(1, state.getRepairKeyspaceCount());
@ -271,7 +266,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetLongestUnrepairedNode()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
AutoRepairHistory history = new AutoRepairHistory(UUID.randomUUID(), "", 0, 0, null, 0, false);
state.setLongestUnrepairedNode(history);
@ -282,7 +277,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetSucceededTokenRangesCount()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setSucceededTokenRangesCount(1);
@ -292,7 +287,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetSucceededTokenRangesCount()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.succeededTokenRangesCount = 1;
assertEquals(1, state.getSucceededTokenRangesCount());
@ -301,7 +296,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testSetFailedTokenRangesCount()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.setFailedTokenRangesCount(1);
@ -311,7 +306,7 @@ public class AutoRepairStateTest extends CQLTester
@Test
public void testGetFailedTokenRangesCount()
{
AutoRepairState state = RepairType.getAutoRepairState(repairType);
AutoRepairState state = RepairType.getAutoRepairState(repairType, new AutoRepairConfig());
state.failedTokenRangesCount = 1;
assertEquals(1, state.getFailedTokenRangesCount());

147
test/unit/org/apache/cassandra/repair/autorepair/FixedSplitTokenRangeSplitterHelper.java
View File

@ -22,8 +22,10 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
@ -39,11 +41,14 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.config.CassandraRelevantProperties.SYSTEM_DISTRIBUTED_DEFAULT_RF;
import static org.apache.cassandra.cql3.CQLTester.Fuzzed.setupSeed;
import static org.apache.cassandra.cql3.CQLTester.Fuzzed.updateConfigs;
import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.split;
import static org.apache.cassandra.repair.autorepair.FixedSplitTokenRangeSplitter.DEFAULT_NUMBER_OF_SUBRANGES;
import static org.apache.cassandra.repair.autorepair.FixedSplitTokenRangeSplitter.NUMBER_OF_SUBRANGES;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@ -58,6 +63,8 @@ public class FixedSplitTokenRangeSplitterHelper
private static final String TABLE2 = "tbl2";
private static final String TABLE3 = "tbl3";
public static final String KEYSPACE = "ks";
public static final List<String> tables = Arrays.asList(TABLE1, TABLE2, TABLE3);
public static final Map<String, Map<Range<Token>, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes = new HashMap<>();
public static void setupClass(int numTokens) throws Exception
{
@ -74,35 +81,59 @@ public class FixedSplitTokenRangeSplitterHelper
SYSTEM_DISTRIBUTED_DEFAULT_RF.setInt(1);
QueryProcessor.executeInternal(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}", FixedSplitTokenRangeSplitterHelper.KEYSPACE));
Pair<Collection<Range<Token>>, Integer> tokensAndWrappedAroundCount = getTokenRangesAndTotalWrapAroundCount();
int totalToken = numTokens + tokensAndWrappedAroundCount.right();
long perTokenSizeTable1 = 512L / totalToken;
long perTokenSizeTable2 = 1024L / totalToken;
long perTokenSizeTable3 = 2048L / totalToken;
for (Range<Token> tokenRange : tokensAndWrappedAroundCount.left)
{
ksTablesEstimatedBytes.put(AutoRepairUtils.getKeyspaceTableName(KEYSPACE, TABLE1), new HashMap<>()
{{
put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, "", "", tokenRange, 0, perTokenSizeTable1, perTokenSizeTable1));
}});
ksTablesEstimatedBytes.put(AutoRepairUtils.getKeyspaceTableName(KEYSPACE, TABLE2), new HashMap<>()
{{
put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, "", "", tokenRange, 0, perTokenSizeTable2, perTokenSizeTable2));
}});
ksTablesEstimatedBytes.put(AutoRepairUtils.getKeyspaceTableName(KEYSPACE, TABLE3), new HashMap<>()
{{
put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, "", "", tokenRange, 0, perTokenSizeTable3, perTokenSizeTable3));
}});
}
}
public static void testTokenRangesSplitByTable(int numTokens, int numberOfSubRanges, AutoRepairConfig.RepairType repairType)
{
int numberOfSplits = calcSplits(numTokens, numberOfSubRanges);
AutoRepairService.instance.getAutoRepairConfig().setRepairByKeyspace(repairType, false);
Collection<Range<Token>> tokens = TokenRingUtils.getPrimaryRangesForEndpoint(KEYSPACE, FBUtilities.getBroadcastAddressAndPort());
AutoRepairConfig config = AutoRepairService.instance.getAutoRepairConfig();
config.setRepairByKeyspace(repairType, false);
Pair<Collection<Range<Token>>, Integer> tokensAndWrappedAroundCount = getTokenRangesAndTotalWrapAroundCount();
Collection<Range<Token>> tokens = tokensAndWrappedAroundCount.left();
// For the test case, the tokens are allocated dynamically, so we do not know which token-ranges wrap around.
// As a result, we need to adjust the token count on a need basis.
numTokens += tokensAndWrappedAroundCount.right();
assertEquals(numTokens, tokens.size());
List<String> tables = Arrays.asList(TABLE1, TABLE2, TABLE3);
List<Range<Token>> expectedToken = new ArrayList<>();
int numberOfSplits = Math.max(1, numberOfSubRanges / tokens.size());
for (int i = 0; i < tables.size(); i++)
{
for (Range<Token> range : tokens)
for (Range<Token> token : tokens)
{
expectedToken.addAll(AutoRepairUtils.split(range, numberOfSplits));
expectedToken.addAll(split(token, numberOfSplits));
}
}
List<PrioritizedRepairPlan> plan = PrioritizedRepairPlan.buildSingleKeyspacePlan(repairType, KEYSPACE, TABLE1, TABLE2, TABLE3);
Iterator<KeyspaceRepairAssignments> keyspaceAssignments = new FixedSplitTokenRangeSplitter(repairType, Collections.singletonMap(FixedSplitTokenRangeSplitter.NUMBER_OF_SUBRANGES, Integer.toString(numberOfSubRanges)))
.getRepairAssignments(true, plan);
Iterator<KeyspaceRepairAssignments> keyspaceAssignments =
new FixedSplitTokenRangeSplitter(repairType, Collections.singletonMap(NUMBER_OF_SUBRANGES, Integer.toString(numberOfSubRanges)))
.getRepairAssignments(config.getRepairPrimaryTokenRangeOnly(repairType), getPlan(repairType));
// should be only 1 entry for the keyspace.
assertTrue(keyspaceAssignments.hasNext());
KeyspaceRepairAssignments keyspace = keyspaceAssignments.next();
KeyspaceRepairAssignments keyspaceRepairAssignment = keyspaceAssignments.next();
assertFalse(keyspaceAssignments.hasNext());
List<RepairAssignment> assignments = keyspace.getRepairAssignments();
List<RepairAssignment> assignments = keyspaceRepairAssignment.getRepairAssignments();
assertEquals(numTokens * numberOfSplits * tables.size(), assignments.size());
assertEquals(expectedToken.size(), assignments.size());
@ -113,9 +144,12 @@ public class FixedSplitTokenRangeSplitterHelper
List<Range<Token>> expectedTokensForATable = new ArrayList<>();
for (int j = 0; j < assignmentsPerTable; j++)
{
assertEquals(Collections.singletonList(tables.get(i)), assignments.get(i * assignmentsPerTable + j).getTableNames());
assignmentForATable.add(assignments.get(i * assignmentsPerTable + j));
expectedTokensForATable.add(expectedToken.get(i * assignmentsPerTable + j));
long expectedBytes = ksTablesEstimatedBytes.get(AutoRepairUtils.getKeyspaceTableName(KEYSPACE, tables.get(i))).values().stream().mapToLong(sizeEstimate -> sizeEstimate.sizeForRepair).sum() / numberOfSplits;
int theTableAssignmentIdx = i * assignmentsPerTable + j;
assertEquals(expectedBytes, assignments.get(theTableAssignmentIdx).estimatedBytes);
assertEquals(Collections.singletonList(tables.get(i)), assignments.get(theTableAssignmentIdx).getTableNames());
assignmentForATable.add(assignments.get(theTableAssignmentIdx));
expectedTokensForATable.add(expectedToken.get(theTableAssignmentIdx));
}
compare(numTokens, numberOfSplits, expectedTokensForATable, assignmentForATable);
}
@ -123,20 +157,24 @@ public class FixedSplitTokenRangeSplitterHelper
public static void testTokenRangesSplitByKeyspace(int numTokens, int numberOfSubRanges, AutoRepairConfig.RepairType repairType)
{
int numberOfSplits = calcSplits(numTokens, numberOfSubRanges);
AutoRepairService.instance.getAutoRepairConfig().setRepairByKeyspace(repairType, true);
Collection<Range<Token>> tokens = TokenRingUtils.getPrimaryRangesForEndpoint(KEYSPACE, FBUtilities.getBroadcastAddressAndPort());
AutoRepairConfig config = AutoRepairService.instance.getAutoRepairConfig();
config.setRepairByKeyspace(repairType, true);
Pair<Collection<Range<Token>>, Integer> tokensAndWrappedRanges = getTokenRangesAndTotalWrapAroundCount();
Collection<Range<Token>> tokens = tokensAndWrappedRanges.left();
// For the test case, the tokens are allocated dynamically, so we do not know which token-ranges wrap around.
// As a result, we need to adjust the token count on a need basis.
numTokens += tokensAndWrappedRanges.right();
assertEquals(numTokens, tokens.size());
int numberOfSplits = Math.max(1, numberOfSubRanges / tokens.size());
List<Range<Token>> expectedToken = new ArrayList<>();
for (Range<Token> range : tokens)
{
expectedToken.addAll(AutoRepairUtils.split(range, numberOfSplits));
}
List<PrioritizedRepairPlan> plan = PrioritizedRepairPlan.buildSingleKeyspacePlan(repairType, KEYSPACE, TABLE1, TABLE2, TABLE3);
Iterator<KeyspaceRepairAssignments> keyspaceAssignments = new FixedSplitTokenRangeSplitter(repairType, Collections.singletonMap(FixedSplitTokenRangeSplitter.NUMBER_OF_SUBRANGES, Integer.toString(numberOfSubRanges)))
.getRepairAssignments(true, plan);
Iterator<KeyspaceRepairAssignments> keyspaceAssignments =
new FixedSplitTokenRangeSplitter(repairType, Collections.singletonMap(NUMBER_OF_SUBRANGES, Integer.toString(numberOfSubRanges)))
.getRepairAssignments(config.getRepairPrimaryTokenRangeOnly(repairType), getPlan(repairType));
// should be only 1 entry for the keyspace.
assertTrue(keyspaceAssignments.hasNext());
@ -150,35 +188,20 @@ public class FixedSplitTokenRangeSplitterHelper
assertEquals(expectedToken.size(), assignments.size());
compare(numTokens, numberOfSplits, expectedToken, assignments);
for (int i = 0; i < assignments.size(); i++)
{
assertEquals(assignments.get(i).estimatedBytes,
ksTablesEstimatedBytes.values().stream()
.flatMap(tableMap -> tableMap.values().stream())
.mapToLong(sizeEstimate -> sizeEstimate.sizeForRepair)
.sum() / numberOfSplits);
}
}
public static void testTokenRangesWithDefaultSplit(int numTokens, AutoRepairConfig.RepairType repairType)
{
int numberOfSplits = calcSplits(numTokens, DEFAULT_NUMBER_OF_SUBRANGES);
Collection<Range<Token>> tokens = TokenRingUtils.getPrimaryRangesForEndpoint(KEYSPACE, FBUtilities.getBroadcastAddressAndPort());
assertEquals(numTokens, tokens.size());
List<Range<Token>> expectedToken = new ArrayList<>();
for (Range<Token> range : tokens)
{
expectedToken.addAll(AutoRepairUtils.split(range, numberOfSplits));
}
List<PrioritizedRepairPlan> plan = PrioritizedRepairPlan.buildSingleKeyspacePlan(repairType, KEYSPACE, TABLE1);
Iterator<KeyspaceRepairAssignments> keyspaceAssignments = new FixedSplitTokenRangeSplitter(repairType, Collections.emptyMap()).getRepairAssignments(true, plan);
// should be only 1 entry for the keyspace.
assertTrue(keyspaceAssignments.hasNext());
KeyspaceRepairAssignments keyspace = keyspaceAssignments.next();
assertFalse(keyspaceAssignments.hasNext());
List<RepairAssignment> assignments = keyspace.getRepairAssignments();
assertNotNull(assignments);
// should be 3 entries for the table which covers each token range.
assertEquals(numTokens * numberOfSplits, assignments.size());
compare(numTokens, numberOfSplits, expectedToken, assignments);
testTokenRangesSplitByKeyspace(numTokens, DEFAULT_NUMBER_OF_SUBRANGES, repairType);
}
private static void compare(int numTokens, int numberOfSplits, List<Range<Token>> expectedToken, List<RepairAssignment> assignments)
@ -194,8 +217,34 @@ public class FixedSplitTokenRangeSplitterHelper
assertEquals(a, b);
}
private static int calcSplits(int numTokens, int subRange)
private static Pair<Collection<Range<Token>>, Integer> getTokenRangesAndTotalWrapAroundCount()
{
return Math.max(1, subRange / numTokens);
int wrappedRanges = 0;
Collection<Range<Token>> ranges = TokenRingUtils.getPrimaryRangesForEndpoint(KEYSPACE, FBUtilities.getBroadcastAddressAndPort());
Collection<Range<Token>> tokens = new ArrayList<>();
for (Range<Token> wrappedRange : ranges)
{
if (wrappedRange.isWrapAround())
{
wrappedRanges++;
}
tokens.addAll(wrappedRange.unwrap());
}
return Pair.create(tokens, wrappedRanges);
}
private static List<PrioritizedRepairPlan> getPlan(AutoRepairConfig.RepairType repairType)
{
AutoRepairConfig config = AutoRepairService.instance.getAutoRepairConfig();
List<PrioritizedRepairPlan> plan = PrioritizedRepairPlan.build(new HashMap<>()
{{
put(KEYSPACE, tables);
}}, repairType, (l) -> {
},
config.getRepairPrimaryTokenRangeOnly(repairType));
assertEquals(1, plan.size());
assertEquals(1, plan.get(0).getKeyspaceRepairPlans().size());
plan.get(0).getKeyspaceRepairPlans().get(0).ksTablesEstimatedBytes = ksTablesEstimatedBytes;
return plan;
}
}

25
test/unit/org/apache/cassandra/repair/autorepair/PrioritizedRepairPlanTest.java
View File

@ -18,14 +18,18 @@
package org.apache.cassandra.repair.autorepair;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.google.common.collect.Lists;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.service.AutoRepairService;
import org.apache.cassandra.service.StorageService;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@ -35,6 +39,13 @@ import static org.junit.Assert.assertTrue;
*/
public class PrioritizedRepairPlanTest extends CQLTester
{
@BeforeClass
public static void setup()
{
StorageService.instance.doAutoRepairSetup();
}
@Test
public void testBuildWithDifferentPriorities()
{
@ -43,7 +54,8 @@ public class PrioritizedRepairPlanTest extends CQLTester
String table2 = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH auto_repair = {'full_enabled': 'true', 'priority': '3'}");
String table3 = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH auto_repair = {'full_enabled': 'true', 'priority': '1'}");
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.buildSingleKeyspacePlan(AutoRepairConfig.RepairType.FULL, KEYSPACE, table1, table2, table3);
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.build(new HashMap<>(){{put(KEYSPACE, Arrays.asList(table1, table2, table3));}}, AutoRepairConfig.RepairType.FULL, (l) -> {},
AutoRepairService.instance.getAutoRepairConfig().getRepairPrimaryTokenRangeOnly(AutoRepairConfig.RepairType.FULL));
assertEquals(3, prioritizedRepairPlans.size());
// Verify the order is by descending priority and matches the expected tables
@ -66,7 +78,8 @@ public class PrioritizedRepairPlanTest extends CQLTester
String table3 = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH auto_repair = {'full_enabled': 'true', 'priority': '2'}");
// Expect only 1 plan since all tables share the same priority
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.buildSingleKeyspacePlan(AutoRepairConfig.RepairType.FULL, KEYSPACE, table1, table2, table3);
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.build(new HashMap<>(){{put(KEYSPACE, Arrays.asList(table1, table2, table3));}}, AutoRepairConfig.RepairType.FULL, (l) -> {},
AutoRepairService.instance.getAutoRepairConfig().getRepairPrimaryTokenRangeOnly(AutoRepairConfig.RepairType.FULL));
assertEquals(1, prioritizedRepairPlans.size());
// Verify all tables present in the plan
@ -101,7 +114,7 @@ public class PrioritizedRepairPlanTest extends CQLTester
keyspaceToTableMap.put(ks2, Lists.newArrayList(table6, table7));
// Expect 4 plans
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.build(keyspaceToTableMap, AutoRepairConfig.RepairType.FULL, java.util.Collections::sort);
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.build(keyspaceToTableMap, AutoRepairConfig.RepairType.FULL, java.util.Collections::sort, true);
assertEquals(4, prioritizedRepairPlans.size());
// Verify the order is by descending priority and matches the expected tables
@ -143,7 +156,8 @@ public class PrioritizedRepairPlanTest extends CQLTester
public void testBuildWithEmptyTableList()
{
// Test with an empty table list (should remain empty)
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.buildSingleKeyspacePlan(AutoRepairConfig.RepairType.FULL, KEYSPACE);
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.build(new HashMap<>(){{put(KEYSPACE, Arrays.asList());}}, AutoRepairConfig.RepairType.FULL, (l) -> {},
AutoRepairService.instance.getAutoRepairConfig().getRepairPrimaryTokenRangeOnly(AutoRepairConfig.RepairType.FULL));
assertTrue(prioritizedRepairPlans.isEmpty());
}
@ -154,7 +168,8 @@ public class PrioritizedRepairPlanTest extends CQLTester
String table1 = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH auto_repair = {'full_enabled': 'true', 'priority': '5'}");
// Expect only 1 plans
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.buildSingleKeyspacePlan(AutoRepairConfig.RepairType.FULL, KEYSPACE, table1);
List<PrioritizedRepairPlan> prioritizedRepairPlans = PrioritizedRepairPlan.build(new HashMap<>(){{put(KEYSPACE, Arrays.asList(table1));}}, AutoRepairConfig.RepairType.FULL, (l) -> {},
AutoRepairService.instance.getAutoRepairConfig().getRepairPrimaryTokenRangeOnly(AutoRepairConfig.RepairType.FULL));
assertEquals(1, prioritizedRepairPlans.size());
// Verify the order is by descending priority and matches the expected tables

196
test/unit/org/apache/cassandra/repair/autorepair/RepairTokenRangeSplitterTest.java
View File

@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@ -39,6 +40,7 @@ import org.apache.cassandra.config.DataStorageSpec.LongMebibytesBound;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
@ -46,11 +48,13 @@ import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.io.sstable.format.bti.BtiFormat;
import org.apache.cassandra.repair.autorepair.AutoRepairConfig.RepairType;
import org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.FilteredRepairAssignments;
import org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.SizeEstimate;
import org.apache.cassandra.repair.autorepair.AutoRepairUtils.SizeEstimate;
import org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.SizedRepairAssignment;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.AutoRepairService;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.apache.cassandra.repair.autorepair.AutoRepairUtils.getKeyspaceTableName;
import static org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.MAX_BYTES_PER_SCHEDULE;
import static org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.BYTES_PER_ASSIGNMENT;
import static org.apache.cassandra.repair.autorepair.RepairTokenRangeSplitter.MAX_TABLES_PER_ASSIGNMENT;
@ -111,7 +115,7 @@ public class RepairTokenRangeSplitterTest extends CQLTester
try (Refs<SSTableReader> sstables = RepairTokenRangeSplitter.getSSTableReaderRefs(RepairType.FULL, KEYSPACE, tableName, FULL_RANGE))
{
assertEquals(10, sstables.iterator().next().getEstimatedPartitionSize().count());
SizeEstimate sizes = RepairTokenRangeSplitter.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, FULL_RANGE, sstables);
SizeEstimate sizes = AutoRepairUtils.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, FULL_RANGE, sstables);
assertEquals(10, sizes.partitions);
}
}
@ -132,8 +136,8 @@ public class RepairTokenRangeSplitterTest extends CQLTester
try (Refs<SSTableReader> sstables1 = RepairTokenRangeSplitter.getSSTableReaderRefs(RepairType.FULL, KEYSPACE, tableName, tokenRange1);
Refs<SSTableReader> sstables2 = RepairTokenRangeSplitter.getSSTableReaderRefs(RepairType.FULL, KEYSPACE, tableName, tokenRange2))
{
SizeEstimate sizes1 = RepairTokenRangeSplitter.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, tokenRange1, sstables1);
SizeEstimate sizes2 = RepairTokenRangeSplitter.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, tokenRange2, sstables2);
SizeEstimate sizes1 = AutoRepairUtils.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, tokenRange1, sstables1);
SizeEstimate sizes2 = AutoRepairUtils.getSizesForRangeOfSSTables(RepairType.FULL, KEYSPACE, tableName, tokenRange2, sstables2);
// +-5% because including entire compression blocks covering token range, HLL merge and the applying of range size approx ratio causes estimation errors
long allowableDelta = (long) (partitionCount * .05);
@ -143,68 +147,150 @@ public class RepairTokenRangeSplitterTest extends CQLTester
}
@Test
public void testGetRepairAssignmentsForTable_NoSSTables()
public void testGetRepairAssignmentsForTableNoSSTables()
{
// Should return 1 assignment if there are no SSTables
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForTable(CQLTester.KEYSPACE, tableName, FULL_RANGE);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForTable(new KeyspaceRepairPlan(CQLTester.KEYSPACE, Collections.singletonList(tableName), AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, Collections.singletonList(tableName), Collections.singletonList(FULL_RANGE))), tableName, FULL_RANGE);
assertEquals(1, assignments.size());
}
@Test
public void testGetRepairAssignmentsForTable_Single()
public void testGetRepairAssignmentsForTableSingle()
{
insertAndFlushSingleTable();
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForTable(CQLTester.KEYSPACE, tableName, FULL_RANGE);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForTable(new KeyspaceRepairPlan(CQLTester.KEYSPACE, Collections.singletonList(tableName), AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, Collections.singletonList(tableName), Collections.singletonList(FULL_RANGE))), tableName, FULL_RANGE);
assertEquals(1, assignments.size());
}
@Test
public void testGetRepairAssignmentsForTable_BatchingTables()
public void testGetRepairAssignmentsForTableBatchingTablesCompressed()
{
repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "2"));
List<String> tableNames = createAndInsertTables(3);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, KEYSPACE, tableNames, FULL_RANGE);
List<String> tableNames = createAndInsertTables(3, true);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE);
// We expect two assignments, one with table1 and table2 batched, and one with table3
assertEquals(2, assignments.size());
assertEquals(2, assignments.get(0).getTableNames().size());
assertEquals(1, assignments.get(1).getTableNames().size());
assertEquals(new HashSet<>(Arrays.asList(tableNames.get(0), tableNames.get(1))), new HashSet<>(assignments.get(0).getTableNames()));
assertEquals(tableNames.get(2), assignments.get(1).getTableNames().get(0));
}
@Test
public void testGetRepairAssignmentsForTable_BatchSize()
public void testGetRepairAssignmentsForTableBatchingTablesUncompressed()
{
repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "2"));
List<String> tableNames = createAndInsertTables(2);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, KEYSPACE, tableNames, FULL_RANGE);
List<String> tableNames = createAndInsertTables(3, false);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE);
// We expect two assignments, one with table1 and table2 batched, and one with table3
assertEquals(2, assignments.size());
assertEquals(2, assignments.get(0).getTableNames().size());
assertEquals(1, assignments.get(1).getTableNames().size());
assertEquals(new HashSet<>(Arrays.asList(tableNames.get(0), tableNames.get(1))), new HashSet<>(assignments.get(0).getTableNames()));
assertEquals(tableNames.get(2), assignments.get(1).getTableNames().get(0));
assertTrue(assignments.get(0).getEstimatedBytes() > 0);
assertTrue(assignments.get(1).getEstimatedBytes() > 0);
assertEquals(calculateSSTableSizeOnDisk(Arrays.asList(tableNames.get(0), tableNames.get(1))), assignments.get(0).getEstimatedBytes());
assertEquals(calculateSSTableSizeOnDisk(Collections.singletonList(tableNames.get(2))), assignments.get(1).getEstimatedBytes());
}
@Test
public void testGetRepairAssignmentsForTableBatchSizeCompressed()
{
repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "2"));
List<String> tableNames = createAndInsertTables(2, true);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE);
// We expect one assignment, with two tables batched
assertEquals(1, assignments.size());
assertEquals(2, assignments.get(0).getTableNames().size());
assertEquals(new HashSet<>(Arrays.asList(tableNames.get(0), tableNames.get(1))), new HashSet<>(assignments.get(0).getTableNames()));
}
@Test
public void testGetRepairAssignmentsForTable_NoBatching()
public void testGetRepairAssignmentsForTableBatchSizeUnCompressed()
{
repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "2"));
List<String> tableNames = createAndInsertTables(2, false);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL,
new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE);
// We expect one assignment, with two tables batched
assertEquals(1, assignments.size());
assertEquals(2, assignments.get(0).getTableNames().size());
assertEquals(new HashSet<>(Arrays.asList(tableNames.get(0), tableNames.get(1))), new HashSet<>(assignments.get(0).getTableNames()));
assertTrue(assignments.get(0).getEstimatedBytes() > 0);
assertEquals(calculateSSTableSizeOnDisk(Arrays.asList(tableNames.get(0), tableNames.get(1))), assignments.get(0).getEstimatedBytes());
}
@Test
public void testGetRepairAssignmentsForTableNoBatchingCompressed()
{
repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "1"));
List<String> tableNames = createAndInsertTables(3);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, KEYSPACE, tableNames, FULL_RANGE);
List<String> tableNames = createAndInsertTables(3, true);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE);
assertEquals(3, assignments.size());
assertEquals(Collections.singletonList(tableNames.get(0)), assignments.get(0).getTableNames());
assertEquals(Collections.singletonList(tableNames.get(1)), assignments.get(1).getTableNames());
assertEquals(Collections.singletonList(tableNames.get(2)), assignments.get(2).getTableNames());
}
@Test
public void testGetRepairAssignmentsForTable_AllBatched()
public void testGetRepairAssignmentsForTableNoBatchingUncompressed()
{
repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "1"));
List<String> tableNames = createAndInsertTables(3, false);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE);
assertEquals(3, assignments.size());
assertEquals(Collections.singletonList(tableNames.get(0)), assignments.get(0).getTableNames());
assertEquals(Collections.singletonList(tableNames.get(1)), assignments.get(1).getTableNames());
assertEquals(Collections.singletonList(tableNames.get(2)), assignments.get(2).getTableNames());
assertTrue(assignments.get(0).getEstimatedBytes() > 0);
assertTrue(assignments.get(1).getEstimatedBytes() > 0);
assertTrue(assignments.get(2).getEstimatedBytes() > 0);
assertEquals(calculateSSTableSizeOnDisk(Collections.singletonList(tableNames.get(0))), assignments.get(0).getEstimatedBytes());
assertEquals(calculateSSTableSizeOnDisk(Collections.singletonList(tableNames.get(1))), assignments.get(1).getEstimatedBytes());
assertEquals(calculateSSTableSizeOnDisk(Collections.singletonList(tableNames.get(2))), assignments.get(2).getEstimatedBytes());
}
@Test
public void testGetRepairAssignmentsForTableAllBatchedCompressed()
{
repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "100"));
List<String> tableNames = createAndInsertTables(5);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, KEYSPACE, tableNames, FULL_RANGE);
List<String> tableNames = createAndInsertTables(5, true);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE);
assertEquals(1, assignments.size());
assertEquals(5, assignments.get(0).getTableNames().size());
assertEquals(new HashSet<>(tableNames),
new HashSet<>(assignments.get(0).getTableNames()));
}
@Test
public void testGetRepairAssignmentsForTableAllBatchedUncompressed()
{
repairRangeSplitter = new RepairTokenRangeSplitter(RepairType.FULL, Collections.singletonMap(MAX_TABLES_PER_ASSIGNMENT, "100"));
List<String> tableNames = createAndInsertTables(5, false);
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignmentsForKeyspace(RepairType.FULL, new KeyspaceRepairPlan(CQLTester.KEYSPACE, tableNames, AutoRepairUtils.calcTotalBytesToBeRepaired(RepairType.FULL, CQLTester.KEYSPACE, tableNames, Collections.singletonList(FULL_RANGE))), FULL_RANGE);
assertEquals(1, assignments.size());
assertEquals(5, assignments.get(0).getTableNames().size());
assertEquals(new HashSet<>(tableNames),
new HashSet<>(assignments.get(0).getTableNames()));
assertTrue(assignments.get(0).getEstimatedBytes() > 0);
assertEquals(calculateSSTableSizeOnDisk(tableNames), assignments.get(0).getEstimatedBytes());
}
@Test(expected = IllegalStateException.class)
@ -312,7 +398,7 @@ public class RepairTokenRangeSplitterTest extends CQLTester
// Given a size estimate of 1024GiB, we should expect 21 splits (50GiB*21 = 1050GiB < 1024GiB)
SizeEstimate sizeEstimate = sizeEstimateByBytes(new LongMebibytesBound("1024GiB"));
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignments(Collections.singletonList(sizeEstimate));
List<SizedRepairAssignment> assignments = repairRangeSplitter.getRepairAssignments(sizeEstimate);
// Should be 21 assignments, each being ~48.76 GiB
assertEquals(21, assignments.size());
@ -335,7 +421,24 @@ public class RepairTokenRangeSplitterTest extends CQLTester
{
AutoRepairService.instance.getAutoRepairConfig().setRepairByKeyspace(RepairType.FULL, true);
final KeyspaceRepairPlan repairPlan = new KeyspaceRepairPlan("system_auth", new ArrayList<>(AuthKeyspace.TABLE_NAMES));
Map<String, Map<Range<Token>, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes = new HashMap<>();
List<Range<Token>> tokenRanges = AutoRepairUtils.getTokenRanges(true, SchemaConstants.AUTH_KEYSPACE_NAME);
long tableSizeInBytes = 100L;
long tableSizeInBytesPerTokenRange = tableSizeInBytes / tokenRanges.size();
for (String tableName : AuthKeyspace.TABLE_NAMES)
{
String ksTableName = getKeyspaceTableName(SchemaConstants.AUTH_KEYSPACE_NAME, tableName);
ksTablesEstimatedBytes.putIfAbsent(ksTableName, new HashMap<>());
Map<Range<Token>, AutoRepairUtils.SizeEstimate> rangeSizeEstimateMap = ksTablesEstimatedBytes.get(ksTableName);
for (Range<Token> tokenRange : tokenRanges)
{
rangeSizeEstimateMap.put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, SchemaConstants.AUTH_KEYSPACE_NAME, tableName, tokenRange, 0, tableSizeInBytesPerTokenRange, tableSizeInBytesPerTokenRange));
}
}
final KeyspaceRepairPlan repairPlan = new KeyspaceRepairPlan(SchemaConstants.AUTH_KEYSPACE_NAME, new ArrayList<>(AuthKeyspace.TABLE_NAMES), ksTablesEstimatedBytes);
assertEquals(tableSizeInBytes * AuthKeyspace.TABLE_NAMES.size(), repairPlan.getEstimatedBytes());
final PrioritizedRepairPlan prioritizedRepairPlan = new PrioritizedRepairPlan(0, List.of(repairPlan));
Iterator<KeyspaceRepairAssignments> keyspaceAssignments = repairRangeSplitter.getRepairAssignments(true, List.of(prioritizedRepairPlan));
@ -363,7 +466,25 @@ public class RepairTokenRangeSplitterTest extends CQLTester
{
AutoRepairService.instance.getAutoRepairConfig().setRepairByKeyspace(RepairType.FULL, false);
final KeyspaceRepairPlan repairPlan = new KeyspaceRepairPlan("system_auth", new ArrayList<>(AuthKeyspace.TABLE_NAMES));
Map<String, Map<Range<Token>, AutoRepairUtils.SizeEstimate>> ksTablesEstimatedBytes = new HashMap<>();
List<Range<Token>> tokenRanges = AutoRepairUtils.getTokenRanges(true, SchemaConstants.AUTH_KEYSPACE_NAME);
long tableSizeInBytes = 100L;
long tableSizeInBytesPerTokenRange = tableSizeInBytes / tokenRanges.size();
for (String tableName : AuthKeyspace.TABLE_NAMES)
{
String ksTableName = getKeyspaceTableName(SchemaConstants.AUTH_KEYSPACE_NAME, tableName);
ksTablesEstimatedBytes.putIfAbsent(ksTableName, new HashMap<>());
Map<Range<Token>, AutoRepairUtils.SizeEstimate> rangeSizeEstimateMap = ksTablesEstimatedBytes.get(ksTableName);
for (Range<Token> tokenRange : tokenRanges)
{
rangeSizeEstimateMap.put(tokenRange, new AutoRepairUtils.SizeEstimate(AutoRepairConfig.RepairType.FULL, SchemaConstants.AUTH_KEYSPACE_NAME, tableName, tokenRange, 0, tableSizeInBytesPerTokenRange, tableSizeInBytesPerTokenRange));
}
}
final KeyspaceRepairPlan repairPlan = new KeyspaceRepairPlan(SchemaConstants.AUTH_KEYSPACE_NAME, new ArrayList<>(AuthKeyspace.TABLE_NAMES), ksTablesEstimatedBytes);
assertEquals(tableSizeInBytes * AuthKeyspace.TABLE_NAMES.size(), repairPlan.getEstimatedBytes());
final PrioritizedRepairPlan prioritizedRepairPlan = new PrioritizedRepairPlan(0, List.of(repairPlan));
Iterator<KeyspaceRepairAssignments> keyspaceAssignments = repairRangeSplitter.getRepairAssignments(true, List.of(prioritizedRepairPlan));
@ -436,18 +557,43 @@ public class RepairTokenRangeSplitterTest extends CQLTester
flush();
}
private List<String> createAndInsertTables(int count)
private List<String> createAndInsertTables(int count, boolean enableCompression)
{
List<String> tableNames = new ArrayList<>();
for (int i = 0; i < count; i++)
{
String tableName = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT)");
String tableName;
if (enableCompression)
{
tableName = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT)");
}
else
{
tableName = createTable("CREATE TABLE %s (k INT PRIMARY KEY, v INT) WITH compression = { 'enabled' : false }");
}
tableNames.add(tableName);
insertAndFlushTable(tableName);
}
return tableNames;
}
private long calculateSSTableSizeOnDisk(List<String> tableNames)
{
long totalSSTableBytes = 0;
for (int i = 0; i < tableNames.size(); i++)
{
String tableName = tableNames.get(i);
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, tableName);
assertNotNull(cfs);
Iterable<SSTableReader> sstables = cfs.getTracker().getView().select(SSTableSet.CANONICAL);
for (SSTableReader sstable : sstables)
{
totalSSTableBytes += sstable.onDiskLength();
}
}
return totalSSTableBytes;
}
private void insertAndFlushTable(String tableName)
{
insertAndFlushTable(tableName, 1);