You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by yc...@apache.org on 2024/01/27 00:20:01 UTC
(cassandra-analytics) branch trunk updated: CASSANDRA-19325 fix range split and use open-closed range notation consistently
This is an automated email from the ASF dual-hosted git repository.
ycai pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/trunk by this push:
new b5ba5fa CASSANDRA-19325 fix range split and use open-closed range notation consistently
b5ba5fa is described below
commit b5ba5fad4df490d1b7d47889361db910589409b8
Author: Yifan Cai <yc...@apache.org>
AuthorDate: Wed Jan 24 14:47:36 2024 -0800
CASSANDRA-19325 fix range split and use open-closed range notation consistently
Patch by Yifan Cai; reviewed by Arjun Ashok, Francisco Guerrero for CASSANDRA-19325
---
CHANGES.txt | 1 +
.../spark/bulkwriter/BulkWriteValidator.java | 6 +-
.../spark/bulkwriter/CassandraClusterInfo.java | 20 +-
.../spark/bulkwriter/CommitCoordinator.java | 6 +-
.../cassandra/spark/bulkwriter/RecordWriter.java | 55 +++---
.../cassandra/spark/bulkwriter/RingInstance.java | 10 +-
.../spark/bulkwriter/SidecarDataTransferApi.java | 2 +-
.../cassandra/spark/bulkwriter/StreamSession.java | 14 +-
.../spark/bulkwriter/TokenPartitioner.java | 32 ++--
.../spark/bulkwriter/token/RangeUtils.java | 148 --------------
.../token/ReplicaAwareFailureHandler.java | 8 +-
.../spark/bulkwriter/token/TokenRangeMapping.java | 12 +-
.../spark/common/model/CassandraInstance.java | 12 +-
.../cassandra/spark/data/PartitionedDataLayer.java | 6 +-
.../spark/data/partitioner/CassandraRing.java | 6 +-
.../spark/data/partitioner/TokenPartitioner.java | 2 +-
.../apache/cassandra/spark/utils/RangeUtils.java | 145 ++++++++------
.../spark/bulkwriter/MockBulkWriterContext.java | 11 +-
.../spark/bulkwriter/RecordWriterTest.java | 212 +++++++++++----------
.../spark/bulkwriter/RingInstanceTest.java | 6 +-
.../bulkwriter/StreamSessionConsistencyTest.java | 6 +-
.../spark/bulkwriter/StreamSessionTest.java | 14 +-
.../spark/bulkwriter/TokenPartitionerTest.java | 105 ++++++----
.../spark/bulkwriter/TokenRangeMappingUtils.java | 52 ++---
.../spark/data/PartitionedDataLayerTests.java | 4 +-
.../spark/data/partitioner/CassandraRingTests.java | 61 +++---
.../data/partitioner/TokenPartitionerTests.java | 10 +-
.../cassandra/spark/utils/RangeUtilsTest.java | 155 ++++++++++-----
.../spark/sparksql/PartitionKeyFilterTests.java | 5 +
.../org/apache/cassandra/bridge/TokenRange.java | 73 ++++---
.../cassandra/spark/data/model/TokenOwner.java | 18 +-
.../spark/data/partitioner/CassandraInstance.java | 3 +-
.../cassandra/spark/reader/IndexDbUtils.java | 9 +-
.../apache/cassandra/spark/reader/IndexReader.java | 8 +-
.../cassandra/spark/reader/IndexDbTests.java | 6 -
.../cassandra/spark/reader/IndexOffsetTests.java | 39 ++--
.../cassandra/spark/reader/SSTableReaderTests.java | 4 +-
37 files changed, 647 insertions(+), 639 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 0845191..88d8100 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
1.0.0
+ * Fix range split and use open-closed range notation consistently (CASSANDRA-19325)
* Add integration tests using in-jvm-dtest to cover blocklisted instances (CASSANDRA-19272)
* Fix bulk writer consistency level validations for blocked instances (CASSANDRA-19257)
* Column type mapping error for timestamp type during bulk writes (CASSANDRA-19223)
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java
index 73c4059..172586f 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java
@@ -99,7 +99,7 @@ public class BulkWriteValidator
LOGGER.warn("[{}]: {} failed on {} with message {}",
uuid,
phase,
- commitResult.instance.getNodeName(),
+ commitResult.instance.nodeName(),
err.errMsg);
failureHandler.addFailure(err.tokenRange, commitResult.instance, err.errMsg);
});
@@ -128,7 +128,7 @@ public class BulkWriteValidator
// If we find any nodes in a totally invalid state, just throw as we can't continue
String errorMessage = String.format("Instance (%s) is in an invalid state (%s) during import. "
+ "Please rerun import once topology changes are complete.",
- instance.getNodeName(), cluster.getInstanceState(instance));
+ instance.nodeName(), cluster.getInstanceState(instance));
throw new RuntimeException(errorMessage);
// Check for blocked instances and ranges for the purpose of logging only.
@@ -141,7 +141,7 @@ public class BulkWriteValidator
.getTokenRanges()
.get(instance);
unavailableRanges.forEach(failedRange -> {
- String nodeDisplayName = instance.getNodeName();
+ String nodeDisplayName = instance.nodeName();
String message = String.format("%s %s", nodeDisplayName, availability.getMessage());
LOGGER.warn("{} failed in phase {} on {} because {}", failedRange, phase, nodeDisplayName, message);
if (shouldAddFailure)
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
index edb5375..c2da22a 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
@@ -108,15 +108,15 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
@Override
public boolean instanceIsAvailable(RingInstance ringInstance)
{
- return instanceIsUp(ringInstance.getRingInstance())
- && instanceIsNormal(ringInstance.getRingInstance())
+ return instanceIsUp(ringInstance.ringInstance())
+ && instanceIsNormal(ringInstance.ringInstance())
&& !instanceIsBlocked(ringInstance);
}
@Override
public InstanceState getInstanceState(RingInstance ringInstance)
{
- return InstanceState.valueOf(ringInstance.getRingInstance().state().toUpperCase());
+ return InstanceState.valueOf(ringInstance.ringInstance().state().toUpperCase());
}
public CassandraContext getCassandraContext()
@@ -207,7 +207,7 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
{
List<SidecarInstance> instances = replicas
.stream()
- .map(replica -> new SidecarInstanceImpl(replica.getNodeName(), getCassandraContext().sidecarPort()))
+ .map(replica -> new SidecarInstanceImpl(replica.nodeName(), getCassandraContext().sidecarPort()))
.collect(Collectors.toList());
return getCassandraContext().getSidecarClient().timeSkew(instances).get();
}
@@ -362,7 +362,7 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
private InstanceAvailability determineInstanceAvailability(RingInstance instance)
{
- if (!instanceIsUp(instance.getRingInstance()))
+ if (!instanceIsUp(instance.ringInstance()))
{
return InstanceAvailability.UNAVAILABLE_DOWN;
}
@@ -370,9 +370,9 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
{
return InstanceAvailability.UNAVAILABLE_BLOCKED;
}
- if (instanceIsNormal(instance.getRingInstance()) ||
- instanceIsTransitioning(instance.getRingInstance()) ||
- instanceIsBeingReplaced(instance.getRingInstance()))
+ if (instanceIsNormal(instance.ringInstance()) ||
+ instanceIsTransitioning(instance.ringInstance()) ||
+ instanceIsBeingReplaced(instance.ringInstance()))
{
return InstanceAvailability.AVAILABLE;
}
@@ -413,7 +413,7 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
.filter(this::instanceIsBlocked)
.collect(Collectors.toSet());
- Set<String> blockedIps = blockedInstances.stream().map(i -> i.getRingInstance().address())
+ Set<String> blockedIps = blockedInstances.stream().map(i -> i.ringInstance().address())
.collect(Collectors.toSet());
// Each token range has hosts by DC. We collate them across all ranges into all hosts by DC
@@ -580,7 +580,7 @@ public class CassandraClusterInfo implements ClusterInfo, Closeable
protected boolean instanceIsBlocked(RingInstance instance)
{
- return conf.blockedInstances.contains(instance.getIpAddress());
+ return conf.blockedInstances.contains(instance.ipAddress());
}
protected boolean instanceIsNormal(RingEntry ringEntry)
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CommitCoordinator.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CommitCoordinator.java
index 0fdadbf..cccfa10 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CommitCoordinator.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CommitCoordinator.java
@@ -132,9 +132,9 @@ public final class CommitCoordinator extends AbstractFuture<List<CommitResult>>
executors.computeIfAbsent(instance,
inst -> MoreExecutors.listeningDecorator(
Executors.newFixedThreadPool(job.getCommitThreadsPerInstance(),
- ThreadUtil.threadFactory("commit-sstable-" + inst.getNodeName()))));
+ ThreadUtil.threadFactory("commit-sstable-" + inst.nodeName()))));
List<String> allUuids = new ArrayList<>(uploadRanges.keySet());
- LOGGER.info("Committing UUIDs={}, Ranges={}, instance={}", allUuids, uploadRanges.values(), instance.getNodeName());
+ LOGGER.info("Committing UUIDs={}, Ranges={}, instance={}", allUuids, uploadRanges.values(), instance.nodeName());
List<List<String>> batches = Lists.partition(allUuids, job.getCommitBatchSize());
return batches.stream().map(uuids -> {
String migrationId = UUID.randomUUID().toString();
@@ -145,7 +145,7 @@ public final class CommitCoordinator extends AbstractFuture<List<CommitResult>>
DataTransferApi.RemoteCommitResult result = transferApi.commitSSTables(instance, migrationId, uuids);
if (result.isSuccess)
{
- LOGGER.info("[{}]: Commit succeeded on {} for {}", migrationId, instance.getNodeName(), uploadRanges);
+ LOGGER.info("[{}]: Commit succeeded on {} for {}", migrationId, instance.nodeName(), uploadRanges);
}
else
{
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java
index ed0a6e9..b9df39d 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RecordWriter.java
@@ -31,11 +31,10 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
@@ -139,16 +138,34 @@ public class RecordWriter implements Serializable
Map<String, Object> valueMap = new HashMap<>();
try
{
- Set<Range<BigInteger>> newRanges = new HashSet<>(initialTokenRangeMapping.getRangeMap().asMapOfRanges().keySet());
+ // preserve the order of ranges
+ Set<Range<BigInteger>> newRanges = new LinkedHashSet<>(initialTokenRangeMapping.getRangeMap()
+ .asMapOfRanges()
+ .keySet());
Range<BigInteger> tokenRange = getTokenRange(taskContext);
- Set<Range<BigInteger>> subRanges = newRanges.contains(tokenRange) ?
- Collections.singleton(tokenRange) :
- getIntersectingSubRanges(newRanges, tokenRange);
+ List<Range<BigInteger>> subRanges = newRanges.contains(tokenRange) ?
+ Collections.singletonList(tokenRange) : // no overlaps
+ getIntersectingSubRanges(newRanges, tokenRange); // has overlaps; split into sub-ranges
+ int currentRangeIndex = 0;
+ Range<BigInteger> currentRange = subRanges.get(currentRangeIndex);
while (dataIterator.hasNext())
{
Tuple2<DecoratedKey, Object[]> rowData = dataIterator.next();
- streamSession = maybeCreateStreamSession(taskContext, streamSession, rowData, subRanges, failureHandler, results);
+ BigInteger token = rowData._1().getToken();
+ // Advance to the next range that contains the token.
+ // The intermediate ranges that do not contain the token will be skipped
+ while (!currentRange.contains(token))
+ {
+ currentRangeIndex++;
+ if (currentRangeIndex >= subRanges.size())
+ {
+ String errMsg = String.format("Received Token %s outside the expected ranges %s", token, subRanges);
+ throw new IllegalStateException(errMsg);
+ }
+ currentRange = subRanges.get(currentRangeIndex);
+ }
+ streamSession = maybeCreateStreamSession(taskContext, streamSession, currentRange, failureHandler, results);
maybeCreateTableWriter(partitionId, baseDir);
writeRow(rowData, valueMap, partitionId, streamSession.getTokenRange());
checkBatchSize(streamSession, partitionId, job);
@@ -205,28 +222,12 @@ public class RecordWriter implements Serializable
*/
private StreamSession maybeCreateStreamSession(TaskContext taskContext,
StreamSession streamSession,
- Tuple2<DecoratedKey, Object[]> rowData,
- Set<Range<BigInteger>> subRanges,
+ Range<BigInteger> currentRange,
ReplicaAwareFailureHandler<RingInstance> failureHandler,
List<StreamResult> results)
throws IOException, ExecutionException, InterruptedException
{
- BigInteger token = rowData._1().getToken();
- Range<BigInteger> tokenRange = getTokenRange(taskContext);
-
- Preconditions.checkState(tokenRange.contains(token),
- String.format("Received Token %s outside of expected range %s", token, tokenRange));
-
- // We have split ranges likely resulting from pending nodes
- // Evaluate creating a new session if the token from current row is part of a sub-range
- if (subRanges.size() > 1)
- {
- // Create session using sub-range that contains the token from current row
- Optional<Range<BigInteger>> matchingSubRangeOpt = subRanges.stream().filter(r -> r.contains(token)).findFirst();
- Preconditions.checkState(matchingSubRangeOpt.isPresent(),
- String.format("Received Token %s outside of expected sub-ranges %s", token, subRanges));
- streamSession = maybeCreateSubRangeSession(taskContext, streamSession, failureHandler, results, matchingSubRangeOpt.get());
- }
+ streamSession = maybeCreateSubRangeSession(taskContext, streamSession, failureHandler, results, currentRange);
// If we do not have any stream session at this point, we create a session using the partition's token range
return (streamSession == null) ? createStreamSession(taskContext) : streamSession;
@@ -267,11 +268,11 @@ public class RecordWriter implements Serializable
/**
* Get ranges from the set that intersect and/or overlap with the provided token range
*/
- private Set<Range<BigInteger>> getIntersectingSubRanges(Set<Range<BigInteger>> ranges, Range<BigInteger> tokenRange)
+ private List<Range<BigInteger>> getIntersectingSubRanges(Set<Range<BigInteger>> ranges, Range<BigInteger> tokenRange)
{
return ranges.stream()
.filter(r -> r.isConnected(tokenRange) && !r.intersection(tokenRange).isEmpty())
- .collect(Collectors.toSet());
+ .collect(Collectors.toList());
}
private boolean haveTokenRangeMappingsChanged(TokenRangeMapping<RingInstance> startTaskMapping, Range<BigInteger> taskTokenRange)
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
index 6d172f1..4aa34b9 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
@@ -54,25 +54,25 @@ public class RingInstance implements CassandraInstance, Serializable
// Used only in tests
@Override
- public String getToken()
+ public String token()
{
return ringEntry.token();
}
@Override
- public String getNodeName()
+ public String nodeName()
{
return ringEntry.fqdn();
}
@Override
- public String getDataCenter()
+ public String datacenter()
{
return ringEntry.datacenter();
}
@Override
- public String getIpAddress()
+ public String ipAddress()
{
return ringEntry.address();
}
@@ -115,7 +115,7 @@ public class RingInstance implements CassandraInstance, Serializable
return ringEntry.toString();
}
- public RingEntry getRingInstance()
+ public RingEntry ringInstance()
{
return ringEntry;
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApi.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApi.java
index 545b010..6ddba7a 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApi.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApi.java
@@ -161,6 +161,6 @@ public class SidecarDataTransferApi implements DataTransferApi
protected SidecarInstanceImpl toSidecarInstance(CassandraInstance instance)
{
- return new SidecarInstanceImpl(instance.getNodeName(), sidecarPort);
+ return new SidecarInstanceImpl(instance.nodeName(), sidecarPort);
}
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java
index 762bde2..044f1a8 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java
@@ -195,7 +195,7 @@ public class StreamSession
private boolean isExclusion(RingInstance ringInstance, Set<RingInstance> failedInstances, Set<String> blockedInstanceIps)
{
return failedInstances.contains(ringInstance)
- || blockedInstanceIps.contains(ringInstance.getIpAddress());
+ || blockedInstanceIps.contains(ringInstance.ipAddress());
}
private void sendSSTables(BulkWriterContext writerContext, SSTableWriter ssTableWriter)
@@ -208,7 +208,7 @@ public class StreamSession
LOGGER.info("[{}]: Pushing SSTable {} to replicas {}",
sessionID, dataFile, replicas.stream()
- .map(RingInstance::getNodeName)
+ .map(RingInstance::nodeName)
.collect(Collectors.joining(",")));
replicas.removeIf(replica -> !trySendSSTableToReplica(writerContext, ssTableWriter, dataFile, ssTableIdx, replica));
}
@@ -250,7 +250,7 @@ public class StreamSession
catch (Exception exception)
{
LOGGER.error("[{}]: Failed to stream range {} to instance {}",
- sessionID, tokenRange, replica.getNodeName(), exception);
+ sessionID, tokenRange, replica.nodeName(), exception);
writerContext.cluster().refreshClusterInfo();
failureHandler.addFailure(tokenRange, replica, exception.getMessage());
errors.add(new StreamError(replica, exception.getMessage()));
@@ -297,7 +297,7 @@ public class StreamSession
{
Preconditions.checkNotNull(fileHash, "All files must have a hash. SSTableWriter should have calculated these. This is a bug.");
long fileSize = Files.size(componentFile);
- LOGGER.info("[{}]: Uploading {} to {}: Size is {}", sessionID, componentFile, instance.getNodeName(), fileSize);
+ LOGGER.info("[{}]: Uploading {} to {}: Size is {}", sessionID, componentFile, instance.nodeName(), fileSize);
writerContext.transfer().uploadSSTableComponent(componentFile, ssTableIdx, instance, sessionID, fileHash);
}
@@ -306,11 +306,11 @@ public class StreamSession
if (writerContext.job().getSkipClean())
{
LOGGER.info("Skip clean requested - not cleaning SSTable session {} on instance {}",
- sessionID, instance.getNodeName());
+ sessionID, instance.nodeName());
return;
}
String jobID = writerContext.job().getId().toString();
- LOGGER.info("Cleaning SSTable session {} on instance {}", sessionID, instance.getNodeName());
+ LOGGER.info("Cleaning SSTable session {} on instance {}", sessionID, instance.nodeName());
try
{
writerContext.transfer().cleanUploadSession(instance, sessionID, jobID);
@@ -318,7 +318,7 @@ public class StreamSession
catch (Exception exception)
{
LOGGER.warn("Failed to clean SSTables on {} for session {} and ignoring errMsg",
- instance.getNodeName(), sessionID, exception);
+ instance.nodeName(), sessionID, exception);
}
}
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TokenPartitioner.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TokenPartitioner.java
index 3bc8992..1e2fc6e 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TokenPartitioner.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/TokenPartitioner.java
@@ -41,7 +41,7 @@ import com.google.common.collect.TreeRangeSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.spark.bulkwriter.token.RangeUtils;
+import org.apache.cassandra.spark.utils.RangeUtils;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.spark.Partitioner;
@@ -58,22 +58,22 @@ public class TokenPartitioner extends Partitioner
private final Integer numberSplits;
public TokenPartitioner(TokenRangeMapping<RingInstance> tokenRangeMapping,
- Integer numberSplits,
+ Integer userSpecifiedNumberSplits,
int defaultParallelism,
Integer cores)
{
- this(tokenRangeMapping, numberSplits, defaultParallelism, cores, true);
+ this(tokenRangeMapping, userSpecifiedNumberSplits, defaultParallelism, cores, true);
}
@VisibleForTesting
public TokenPartitioner(TokenRangeMapping<RingInstance> tokenRangeMapping,
- Integer numberSplits,
+ Integer userSpecifiedNumberSplits,
int defaultParallelism,
Integer cores,
boolean randomize)
{
this.tokenRangeMapping = tokenRangeMapping;
- this.numberSplits = calculateSplits(tokenRangeMapping, numberSplits, defaultParallelism, cores);
+ this.numberSplits = calculateSplits(tokenRangeMapping, userSpecifiedNumberSplits, defaultParallelism, cores);
setupTokenRangeMap(randomize);
validate(); // Intentionally not keeping this in readObject(), it is enough to validate in constructor alone
LOGGER.info("Partition map " + partitionMap);
@@ -87,13 +87,17 @@ public class TokenPartitioner extends Partitioner
return nrPartitions;
}
+ /**
+ * @param key the decorated key
+ * @return the partition (non-negative) for the given key; if key is not present in the partition map, 0 is returned
+ */
@SuppressWarnings("ConstantConditions")
@Override
public int getPartition(Object key)
{
DecoratedKey decoratedKey = (DecoratedKey) key;
-
- return partitionMap.get(decoratedKey.getToken());
+ Integer partition = partitionMap.get(decoratedKey.getToken());
+ return partition == null ? 0 : partition;
}
public int numSplits()
@@ -111,13 +115,13 @@ public class TokenPartitioner extends Partitioner
partitionMap = TreeRangeMap.create();
reversePartitionMap = new HashMap<>();
- final AtomicInteger nextPartitionId = new AtomicInteger(0);
- final List<Range<BigInteger>> subRanges = tokenRangeMapping.getRangeMap()
- .asMapOfRanges()
- .keySet()
- .stream()
- .flatMap(tr -> RangeUtils.split(tr, numberSplits).stream())
- .collect(Collectors.toList());
+ AtomicInteger nextPartitionId = new AtomicInteger(0);
+ List<Range<BigInteger>> subRanges = tokenRangeMapping.getRangeMap()
+ .asMapOfRanges()
+ .keySet()
+ .stream()
+ .flatMap(tr -> RangeUtils.split(tr, numberSplits).stream())
+ .collect(Collectors.toList());
if (randomize)
{
// In order to help distribute the upload load more evenly, shuffle the subranges before assigning a partition
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/RangeUtils.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/RangeUtils.java
deleted file mode 100644
index 733910d..0000000
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/RangeUtils.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * 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.spark.bulkwriter.token;
-
-import java.math.BigInteger;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.BoundType;
-import com.google.common.collect.Multimap;
-import com.google.common.collect.Range;
-
-import org.apache.cassandra.spark.common.model.CassandraInstance;
-import org.apache.cassandra.spark.data.partitioner.Partitioner;
-
-/**
- * Common Cassandra range operations on Guava ranges. Assumes ranges are not wrapped around.
- * It's the responsibility of caller to unwrap ranges. For example, [100, 1] should become
- * [100, MAX] and [MIN, 1]. MIN and MAX depend on {@link Partitioner}.
- */
-public final class RangeUtils
-{
- private RangeUtils()
- {
- }
-
- public static BigInteger sizeOf(Range<BigInteger> range)
- {
- Preconditions.checkArgument(range.lowerEndpoint().compareTo(range.upperEndpoint()) <= 0,
- "RangeUtils assume ranges are not wrap-around");
-
- if (range.isEmpty())
- {
- return BigInteger.ZERO;
- }
-
- BigInteger size = range.upperEndpoint().subtract(range.lowerEndpoint()).add(BigInteger.ONE);
-
- if (range.lowerBoundType() == BoundType.OPEN)
- {
- size = size.subtract(BigInteger.ONE);
- }
-
- if (range.upperBoundType() == BoundType.OPEN)
- {
- size = size.subtract(BigInteger.ONE);
- }
-
- return size;
- }
-
- /**
- * Splits the given range into equal sized small ranges. Number of splits can be controlled by
- * nrSplits. If nrSplits are smaller than size of the range, split size would be set to 1.
- *
- * This is best effort scheme, nrSplits not necessarily as promised and not all splits may not be
- * exact same size.
- *
- * @param range the range to split
- * @param nrSplits the number of sub-ranges into which the range should be divided
- * @return a list of sub-ranges
- */
- public static List<Range<BigInteger>> split(Range<BigInteger> range, int nrSplits)
- {
- Preconditions.checkArgument(range.lowerEndpoint().compareTo(range.upperEndpoint()) <= 0,
- "RangeUtils assume ranges are not wrap-around");
-
- if (range.isEmpty())
- {
- return Collections.emptyList();
- }
-
- Preconditions.checkArgument(nrSplits >= 1, "nrSplits must be greater than or equal to 1");
-
- // Make sure split size is not 0
- BigInteger splitSize = sizeOf(range).divide(BigInteger.valueOf(nrSplits));
- if (splitSize.compareTo(BigInteger.ZERO) == 0)
- {
- splitSize = BigInteger.ONE;
- }
-
- // Start from range lower endpoint and spit ranges of size splitSize, until we cross the range
- BigInteger nextLowerEndpoint = range.lowerBoundType() == BoundType.CLOSED
- ? range.lowerEndpoint()
- : range.lowerEndpoint().add(BigInteger.ONE);
- List<Range<BigInteger>> splits = new ArrayList<>();
- while (range.contains(nextLowerEndpoint))
- {
- BigInteger upperEndpoint = nextLowerEndpoint.add(splitSize);
- splits.add(range.intersection(Range.closedOpen(nextLowerEndpoint, upperEndpoint)));
- nextLowerEndpoint = upperEndpoint;
- }
-
- return splits;
- }
-
- public static <Instance extends CassandraInstance> Multimap<Instance, Range<BigInteger>> calculateTokenRanges(
- List<Instance> instances,
- int replicationFactor,
- Partitioner partitioner)
- {
- Preconditions.checkArgument(replicationFactor != 0, "Calculation token ranges wouldn't work with RF 0");
- Preconditions.checkArgument(instances.size() == 0 || replicationFactor <= instances.size(),
- "Calculation token ranges wouldn't work when RF (" + replicationFactor
- + ") is greater than number of Cassandra instances " + instances.size());
- Multimap<Instance, Range<BigInteger>> tokenRanges = ArrayListMultimap.create();
- for (int index = 0; index < instances.size(); index++)
- {
- Instance instance = instances.get(index);
- int disjointReplica = ((instances.size() + index) - replicationFactor) % instances.size();
- BigInteger rangeStart = new BigInteger(instances.get(disjointReplica).getToken());
- BigInteger rangeEnd = new BigInteger(instance.getToken());
-
- // If start token is greater than or equal to end token we are looking at a wrap around range, split it
- if (rangeStart.compareTo(rangeEnd) >= 0)
- {
- tokenRanges.put(instance, Range.range(rangeStart, BoundType.OPEN, partitioner.maxToken(), BoundType.CLOSED));
- tokenRanges.put(instance, Range.range(partitioner.minToken(), BoundType.OPEN, rangeEnd, BoundType.CLOSED));
- }
- else
- {
- tokenRanges.put(instance, Range.openClosed(rangeStart, rangeEnd));
- }
- }
-
- return tokenRanges;
- }
-}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/ReplicaAwareFailureHandler.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/ReplicaAwareFailureHandler.java
index bbcb7c7..d66f65c 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/ReplicaAwareFailureHandler.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/ReplicaAwareFailureHandler.java
@@ -46,7 +46,7 @@ public class ReplicaAwareFailureHandler<Instance extends CassandraInstance>
public ReplicaAwareFailureHandler(Partitioner partitioner)
{
- failedRangesMap.put(Range.closed(partitioner.minToken(), partitioner.maxToken()), ArrayListMultimap.create());
+ failedRangesMap.put(Range.openClosed(partitioner.minToken(), partitioner.maxToken()), ArrayListMultimap.create());
}
/**
@@ -135,7 +135,7 @@ public class ReplicaAwareFailureHandler<Instance extends CassandraInstance>
boolean isConsistencyLevelMet = true;
Set<String> failedInstanceIPs = failedInstances.stream()
- .map(CassandraInstance::getIpAddress)
+ .map(CassandraInstance::ipAddress)
.collect(Collectors.toSet());
Set<String> datacenters = Collections.emptySet();
ReplicationFactor replicationFactor = tokenRangeMapping.replicationFactor();
@@ -158,8 +158,8 @@ public class ReplicaAwareFailureHandler<Instance extends CassandraInstance>
for (String dc : datacenters)
{
Set<String> failedIpsPerDC = failedInstances.stream()
- .filter(inst -> inst.getDataCenter().matches(dc))
- .map(CassandraInstance::getIpAddress)
+ .filter(inst -> inst.datacenter().matches(dc))
+ .map(CassandraInstance::ipAddress)
.collect(Collectors.toSet());
Set<String> dcWriteReplicas = maybeUpdateWriteReplicasForReplacements(tokenRangeMapping.getWriteReplicas(dc),
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMapping.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMapping.java
index d7d582c..2ffd56d 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMapping.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/token/TokenRangeMapping.java
@@ -152,15 +152,15 @@ public class TokenRangeMapping<Instance extends CassandraInstance> implements Se
public Set<String> getBlockedInstances()
{
return blockedInstances.stream()
- .map(RingInstance::getIpAddress)
+ .map(RingInstance::ipAddress)
.collect(Collectors.toSet());
}
public Set<String> getBlockedInstances(String datacenter)
{
return blockedInstances.stream()
- .filter(r -> r.getDataCenter().equalsIgnoreCase(datacenter))
- .map(RingInstance::getIpAddress)
+ .filter(r -> r.datacenter().equalsIgnoreCase(datacenter))
+ .map(RingInstance::ipAddress)
.collect(Collectors.toSet());
}
@@ -168,15 +168,15 @@ public class TokenRangeMapping<Instance extends CassandraInstance> implements Se
public Set<String> getReplacementInstances()
{
return replacementInstances.stream()
- .map(RingInstance::getIpAddress)
+ .map(RingInstance::ipAddress)
.collect(Collectors.toSet());
}
public Set<String> getReplacementInstances(String datacenter)
{
return replacementInstances.stream()
- .filter(r -> r.getDataCenter().equalsIgnoreCase(datacenter))
- .map(RingInstance::getIpAddress)
+ .filter(r -> r.datacenter().equalsIgnoreCase(datacenter))
+ .map(RingInstance::ipAddress)
.collect(Collectors.toSet());
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java
index 0494b6f..13c465a 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java
@@ -19,13 +19,13 @@
package org.apache.cassandra.spark.common.model;
-public interface CassandraInstance
-{
- String getToken();
+import org.apache.cassandra.spark.data.model.TokenOwner;
- String getNodeName();
+public interface CassandraInstance extends TokenOwner
+{
+ String nodeName();
- String getDataCenter();
+ String datacenter();
- String getIpAddress();
+ String ipAddress();
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/PartitionedDataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/PartitionedDataLayer.java
index 143823a..ccb9f3d 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/PartitionedDataLayer.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/PartitionedDataLayer.java
@@ -43,6 +43,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.spark.utils.RangeUtils;
import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
import org.apache.cassandra.spark.data.partitioner.CassandraRing;
import org.apache.cassandra.spark.data.partitioner.ConsistencyLevel;
@@ -55,7 +56,6 @@ import org.apache.cassandra.spark.sparksql.NoMatchFoundException;
import org.apache.cassandra.spark.sparksql.filters.PartitionKeyFilter;
import org.apache.cassandra.spark.sparksql.filters.SparkRangeFilter;
import org.apache.cassandra.spark.stats.Stats;
-import org.apache.cassandra.spark.utils.RangeUtils;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -228,8 +228,8 @@ public abstract class PartitionedDataLayer extends DataLayer
if (!filters.isEmpty() && filtersInRange.isEmpty())
{
- LOGGER.info("None of the partition key filters overlap with Spark partition token range firstToken={} lastToken{}",
- sparkTokenRange.lowerEndpoint(), sparkTokenRange.upperEndpoint());
+ LOGGER.info("None of the partition key filters overlap with Spark partition token range firstToken={} lastToken={}",
+ sparkTokenRange.firstEnclosedValue(), sparkTokenRange.upperEndpoint());
throw new NoMatchFoundException();
}
return filterNonIntersectingSSTables() ? filtersInRange : filters;
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
index dfc684d..ef9c6ff 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java
@@ -47,8 +47,8 @@ import org.slf4j.LoggerFactory;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
-import org.apache.cassandra.spark.data.ReplicationFactor;
import org.apache.cassandra.spark.utils.RangeUtils;
+import org.apache.cassandra.spark.data.ReplicationFactor;
import static org.apache.cassandra.spark.data.ReplicationFactor.ReplicationStrategy.SimpleStrategy;
@@ -72,7 +72,7 @@ public class CassandraRing implements Serializable
private Partitioner partitioner;
private String keyspace;
private ReplicationFactor replicationFactor;
- private ArrayList<CassandraInstance> instances;
+ private List<CassandraInstance> instances;
private transient RangeMap<BigInteger, List<CassandraInstance>> replicas;
private transient Multimap<CassandraInstance, Range<BigInteger>> tokenRangeMap;
@@ -155,7 +155,7 @@ public class CassandraRing implements Serializable
}
// Calculate token range to replica mapping
- replicas.put(Range.closed(partitioner.minToken(), partitioner.maxToken()), Collections.emptyList());
+ replicas.put(Range.openClosed(partitioner.minToken(), partitioner.maxToken()), Collections.emptyList());
tokenRangeMap.asMap().forEach((instance, ranges) -> ranges.forEach(range -> addReplica(instance, range, replicas)));
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/partitioner/TokenPartitioner.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/partitioner/TokenPartitioner.java
index 158c97b..05a6c3e 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/partitioner/TokenPartitioner.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/partitioner/TokenPartitioner.java
@@ -46,8 +46,8 @@ import org.slf4j.LoggerFactory;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
-import org.apache.cassandra.spark.utils.ByteBufferUtils;
import org.apache.cassandra.spark.utils.RangeUtils;
+import org.apache.cassandra.spark.utils.ByteBufferUtils;
/**
* Util class for partitioning Spark workers across the token ring
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/utils/RangeUtils.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/utils/RangeUtils.java
index 86e391a..423b9dc 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/utils/RangeUtils.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/utils/RangeUtils.java
@@ -31,109 +31,137 @@ import com.google.common.collect.Multimap;
import com.google.common.collect.Range;
import org.apache.cassandra.bridge.TokenRange;
-import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+import org.apache.cassandra.spark.data.model.TokenOwner;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
import org.jetbrains.annotations.NotNull;
/**
* Common Cassandra range operations on Guava ranges. Assumes ranges are not wrapped around.
- * It's the responsibility of caller to unwrap ranges. For example, [100, 1] should become
- * [100, MAX] and [MIN, 1]. MIN and MAX depend on {@link Partitioner}.
+ * It's the responsibility of caller to unwrap ranges. For example, {@code (100, 1]} should become
+ * {@code (100, MAX]} and {@code (MIN, 1]}. MIN and MAX values depend on {@link Partitioner}.
*/
public final class RangeUtils
{
private RangeUtils()
{
- throw new IllegalStateException(getClass() + " is static utility class and shall not be instantiated");
}
- @NotNull
- private static BigInteger sizeOf(@NotNull Range<BigInteger> range)
+ public static BigInteger sizeOf(Range<BigInteger> range)
{
Preconditions.checkArgument(range.lowerEndpoint().compareTo(range.upperEndpoint()) <= 0,
"RangeUtils assume ranges are not wrap-around");
+ Preconditions.checkArgument(range.lowerBoundType() == BoundType.OPEN
+ && range.upperBoundType() == BoundType.CLOSED,
+ "Input must be an open-closed range");
if (range.isEmpty())
{
return BigInteger.ZERO;
}
- BigInteger size = range.upperEndpoint().subtract(range.lowerEndpoint()).add(BigInteger.ONE);
-
- if (range.lowerBoundType() == BoundType.OPEN)
- {
- size = size.subtract(BigInteger.ONE);
- }
-
- if (range.upperBoundType() == BoundType.OPEN)
- {
- size = size.subtract(BigInteger.ONE);
- }
-
- return size;
+ return range.upperEndpoint().subtract(range.lowerEndpoint());
}
/**
- * Splits the given range into equal sized small ranges. Number of splits can be controlled by
- * numberOfSplits. If numberOfSplits are smaller than size of the range, split size would be set to 1.
+ * Splits the given range into equal-sized small ranges. Number of splits can be controlled by
+ * nrSplits. If nrSplits are smaller than size of the range, split size would be set to 1, which is
+ * the minimum allowed. For example, if the input range is {@code (0, 1]} and nrSplits is 10, the split
+ * process yields a single range. Because {@code (0, 1]} cannot be split further.
*
- * This is best effort scheme, numberOfSplits not necessarily as promised and not all splits may not be
- * exact same size.
+ * This is a best-effort scheme. The nrSplits is not necessarily as promised, and not all splits may be
+ * the exact same size.
*
- * @param range the range to split
- * @param numberOfSplits the desired number of splits
- * @return the list of split ranges
+ * @param range the range to split
+ * @param nrSplits the number of sub-ranges into which the range should be divided
+ * @return a list of sub-ranges
*/
- @NotNull
- public static List<Range<BigInteger>> split(@NotNull Range<BigInteger> range, int numberOfSplits)
+ public static List<Range<BigInteger>> split(Range<BigInteger> range, int nrSplits)
{
Preconditions.checkArgument(range.lowerEndpoint().compareTo(range.upperEndpoint()) <= 0,
"RangeUtils assume ranges are not wrap-around");
+ Preconditions.checkArgument(range.lowerBoundType() == BoundType.OPEN
+ && range.upperBoundType() == BoundType.CLOSED,
+ "Input must be an open-closed range");
if (range.isEmpty())
{
return Collections.emptyList();
}
- // Make sure split size is not 0
- BigInteger splitSize = sizeOf(range).divide(BigInteger.valueOf(numberOfSplits)).max(BigInteger.ONE);
+ if (nrSplits == 1 || sizeOf(range).equals(BigInteger.ONE))
+ {
+ // no split required; exit early
+ return Collections.singletonList(range);
+ }
- // Start from range lower endpoint and spit ranges of size splitSize, until we cross the range
- BigInteger nextLowerEndpoint = range.lowerBoundType() == BoundType.CLOSED ? range.lowerEndpoint() : range.lowerEndpoint().add(BigInteger.ONE);
+ Preconditions.checkArgument(nrSplits >= 1, "nrSplits must be greater than or equal to 1");
+
+ // The following algorithm tries to get a more evenly-split ranges.
+ // For example, if we split the range (0, 11] into 4 sub-ranges, the naive split yields the following:
+ // (0, 2], (2, 4], (4, 6], (6, 11]
+ // As you can see, the last range is significantly larger than the prior range.
+ // The desired split is (0, 3], (3, 6], (6, 9], (9, 11]. The sizes of the ranges are more close to each other.
+ // --
+ // The procedure of the algorithm is as simple as the below 2 steps:
+ // 1. Get the quotient and the remainder from the integer division.
+ // 2. For each range, as long as the remainder is not 0, we move 1 from the remainder to it.
+ // See org.apache.cassandra.spark.utils.RangeUtilsTest.testSplitYieldMoreEvenRanges
+ // Given that the remainder is always smaller than the divisor (nrSplits), the remainder is exhausted
+ // before the for-loop end.
+ // In some special cases, for example, split (0, 3] into 5 sub-ranges. The quotient is 0 and the remainder is 3.
+ // When the remainder is exhausted, the input is also exhausted. It yields, (0, 1], (1, 2], (2, 3].
+ // See org.apache.cassandra.spark.utils.RangeUtilsTest.testSplitNotSatisfyNrSplits
+ BigInteger[] divideAndRemainder = sizeOf(range).divideAndRemainder(BigInteger.valueOf(nrSplits));
+ BigInteger quotient = divideAndRemainder[0]; // quotient could be 0
+ int remainder = divideAndRemainder[1].intValue(); // remainder must be smaller than nrSplit, which is an integer
+ BigInteger lowerEndpoint = range.lowerEndpoint();
List<Range<BigInteger>> splits = new ArrayList<>();
- while (range.contains(nextLowerEndpoint))
+ for (int i = 0; i < nrSplits; i++)
{
- BigInteger upperEndpoint = nextLowerEndpoint.add(splitSize);
- splits.add(range.intersection(Range.closedOpen(nextLowerEndpoint, upperEndpoint)));
- nextLowerEndpoint = upperEndpoint;
+ BigInteger upperEndpoint = lowerEndpoint.add(quotient);
+ if (remainder > 0)
+ {
+ upperEndpoint = upperEndpoint.add(BigInteger.ONE);
+ remainder--;
+ }
+ if (i + 1 == nrSplits || upperEndpoint.compareTo(range.upperEndpoint()) >= 0)
+ {
+ splits.add(Range.openClosed(lowerEndpoint, range.upperEndpoint()));
+ break; // the split process terminate early because the original range is exhausted
+ }
+ splits.add(Range.openClosed(lowerEndpoint, upperEndpoint));
+ lowerEndpoint = upperEndpoint;
}
return splits;
}
- @NotNull
- public static Multimap<CassandraInstance, Range<BigInteger>> calculateTokenRanges(
- @NotNull List<CassandraInstance> instances,
- int replicationFactor,
- @NotNull Partitioner partitioner)
+ public static <Instance extends TokenOwner> Multimap<Instance, Range<BigInteger>>
+ calculateTokenRanges(List<Instance> instances,
+ int replicationFactor,
+ Partitioner partitioner)
{
- Preconditions.checkArgument(replicationFactor != 0, "RF cannot be 0");
- Preconditions.checkArgument(instances.size() == 0 || replicationFactor <= instances.size(),
- String.format("RF (%d) cannot be greater than the number of Cassandra instances (%d)",
- replicationFactor, instances.size()));
- Multimap<CassandraInstance, Range<BigInteger>> tokenRanges = ArrayListMultimap.create();
+ Preconditions.checkArgument(replicationFactor != 0, "Calculation token ranges wouldn't work with RF 0");
+ Preconditions.checkArgument(instances.isEmpty() || replicationFactor <= instances.size(),
+ "Calculation token ranges wouldn't work when RF (" + replicationFactor
+ + ") is greater than number of Cassandra instances " + instances.size());
+ Multimap<Instance, Range<BigInteger>> tokenRanges = ArrayListMultimap.create();
for (int index = 0; index < instances.size(); index++)
{
- CassandraInstance instance = instances.get(index);
+ Instance instance = instances.get(index);
int disjointReplica = ((instances.size() + index) - replicationFactor) % instances.size();
BigInteger rangeStart = new BigInteger(instances.get(disjointReplica).token());
BigInteger rangeEnd = new BigInteger(instance.token());
- // If start token is not strictly smaller than end token we are looking at a wrap around range, split it
+ // If start token is greater than or equal to end token we are looking at a wrap around range, split it
if (rangeStart.compareTo(rangeEnd) >= 0)
{
- tokenRanges.put(instance, Range.range(rangeStart, BoundType.OPEN, partitioner.maxToken(), BoundType.CLOSED));
- tokenRanges.put(instance, Range.range(partitioner.minToken(), BoundType.CLOSED, rangeEnd, BoundType.CLOSED));
+ tokenRanges.put(instance, Range.openClosed(rangeStart, partitioner.maxToken()));
+ // Skip adding the empty range (minToken, minToken]
+ if (!rangeEnd.equals(partitioner.minToken()))
+ {
+ tokenRanges.put(instance, Range.openClosed(partitioner.minToken(), rangeEnd));
+ }
}
else
{
@@ -147,22 +175,15 @@ public final class RangeUtils
@NotNull
public static TokenRange toTokenRange(@NotNull Range<BigInteger> range)
{
- BigInteger lowerEndpoint = range.lowerEndpoint();
- if (range.lowerBoundType() == BoundType.OPEN)
- {
- lowerEndpoint = lowerEndpoint.add(BigInteger.ONE);
- }
- BigInteger upperEndpoint = range.upperEndpoint();
- if (range.upperBoundType() == BoundType.OPEN)
- {
- upperEndpoint = upperEndpoint.subtract(BigInteger.ONE);
- }
- return TokenRange.closed(lowerEndpoint, upperEndpoint);
+ Preconditions.checkArgument(range.lowerBoundType() == BoundType.OPEN
+ && range.upperBoundType() == BoundType.CLOSED,
+ "Input must be an open-closed range");
+ return TokenRange.openClosed(range.lowerEndpoint(), range.upperEndpoint());
}
@NotNull
public static Range<BigInteger> fromTokenRange(@NotNull TokenRange range)
{
- return Range.closed(range.lowerEndpoint(), range.upperEndpoint());
+ return Range.openClosed(range.lowerEndpoint(), range.upperEndpoint());
}
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java
index 2d8d889..1f589da 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java
@@ -77,6 +77,7 @@ public class MockBulkWriterContext implements BulkWriterContext, ClusterInfo, Jo
private RowBufferMode rowBufferMode = RowBufferMode.UNBUFFERED;
private ConsistencyLevel.CL consistencyLevel;
private int sstableDataSizeInMB = 128;
+ private int sstableWriteBatchSize = 2;
public interface CommitResultSupplier extends BiFunction<List<String>, String, RemoteCommitResult>
{
@@ -245,7 +246,13 @@ public class MockBulkWriterContext implements BulkWriterContext, ClusterInfo, Jo
@Override
public int getSstableBatchSize()
{
- return 2;
+ return sstableWriteBatchSize;
+ }
+
+ @VisibleForTesting
+ void setSstableWriteBatchSize(int sstableWriteBatchSize)
+ {
+ this.sstableWriteBatchSize = sstableWriteBatchSize;
}
@Override
@@ -324,7 +331,7 @@ public class MockBulkWriterContext implements BulkWriterContext, ClusterInfo, Jo
public RemoteCommitResult commitSSTables(CassandraInstance instance, String migrationId, List<String> uuids)
{
commits.computeIfAbsent(instance, k -> new ArrayList<>()).add(migrationId);
- return crSupplier.apply(buildCompleteBatchIds(uuids), instance.getDataCenter());
+ return crSupplier.apply(buildCompleteBatchIds(uuids), instance.datacenter());
}
private List<String> buildCompleteBatchIds(List<String> uuids)
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
index 1d06c42..61c6c2e 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java
@@ -21,7 +21,7 @@ package org.apache.cassandra.spark.bulkwriter;
import java.math.BigInteger;
import java.nio.file.Path;
-import java.util.ArrayList;
+import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -34,22 +34,21 @@ import java.util.stream.IntStream;
import java.util.stream.Stream;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterators;
import com.google.common.collect.Range;
import org.apache.commons.lang3.tuple.Pair;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-import org.apache.cassandra.bridge.RowBufferMode;
import org.apache.cassandra.spark.bulkwriter.token.ConsistencyLevel;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.common.model.CassandraInstance;
-import org.mockito.Mockito;
import org.apache.cassandra.spark.data.CqlField;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
+import org.mockito.Mockito;
import scala.Tuple2;
import static org.apache.cassandra.spark.bulkwriter.MockBulkWriterContext.DEFAULT_CASSANDRA_VERSION;
@@ -58,9 +57,9 @@ import static org.apache.cassandra.spark.bulkwriter.SqlToCqlTypeConverter.INT;
import static org.apache.cassandra.spark.bulkwriter.SqlToCqlTypeConverter.VARCHAR;
import static org.apache.cassandra.spark.bulkwriter.TableSchemaTestCommon.mockCqlType;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.endsWith;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.startsWith;
-import static org.hamcrest.Matchers.endsWith;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -72,9 +71,12 @@ import static org.mockito.Mockito.when;
public class RecordWriterTest
{
- public static final int REPLICA_COUNT = 3;
- public static final int FILES_PER_SSTABLE = 8;
- public static final int UPLOADED_TABLES = 3;
+ private static final int REPLICA_COUNT = 3;
+ private static final int FILES_PER_SSTABLE = 8;
+ // writing 5 rows with batch size of 2 should produce 3 sstable
+ private static final int UPLOADED_SSTABLES = 3;
+ private static final int ROWS_COUNT = 5;
+ private static final int BATCH_SIZE = 2;
private static final String[] COLUMN_NAMES = {
"id", "date", "course", "marks"
};
@@ -96,6 +98,7 @@ public class RecordWriterTest
tw = new MockTableWriter(folder.getRoot());
tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(100000, ImmutableMap.of("DC1", 3), 12);
writerContext = new MockBulkWriterContext(tokenRangeMapping);
+ writerContext.setSstableWriteBatchSize(BATCH_SIZE); // create a new sstable after writing 2 rows
tc = new TestTaskContext();
range = writerContext.job().getTokenPartitioner().getTokenRange(tc.partitionId());
tokenizer = new Tokenizer(writerContext);
@@ -118,7 +121,7 @@ public class RecordWriterTest
rw = new RecordWriter(m, COLUMN_NAMES, () -> tc, SSTableWriter::new);
when(m.getTokenRangeMapping(false)).thenCallRealMethod().thenReturn(testMapping);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
RuntimeException ex = assertThrows(RuntimeException.class, () -> rw.write(data));
assertThat(ex.getMessage(), endsWith("Token range mappings have changed since the task started"));
}
@@ -138,7 +141,7 @@ public class RecordWriterTest
Map<RingInstance, InstanceAvailability> availability
= instances.stream()
.collect(Collectors.toMap(Function.identity(),
- i -> (i.getIpAddress().equals(blockedInstanceIp)) ?
+ i -> (i.ipAddress().equals(blockedInstanceIp)) ?
InstanceAvailability.UNAVAILABLE_BLOCKED :
InstanceAvailability.AVAILABLE));
@@ -148,12 +151,12 @@ public class RecordWriterTest
when(m.getTokenRangeMapping(anyBoolean())).thenReturn(testMapping);
when(m.getInstanceAvailability()).thenReturn(availability);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
rw.write(data);
Map<CassandraInstance, List<UploadRequest>> uploads = writerContext.getUploads();
// Should not upload to blocked instances
assertThat(uploads.keySet().size(), is(REPLICA_COUNT - 1));
- assertFalse(uploads.keySet().stream().map(i -> i.getIpAddress()).collect(Collectors.toSet()).contains(blockedInstanceIp));
+ assertFalse(uploads.keySet().stream().map(i -> i.ipAddress()).collect(Collectors.toSet()).contains(blockedInstanceIp));
}
@Test
@@ -169,7 +172,7 @@ public class RecordWriterTest
when(m.getTokenRangeMapping(anyBoolean())).thenReturn(testMapping);
when(m.getInstanceAvailability()).thenCallRealMethod();
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
rw.write(data);
Map<CassandraInstance, List<UploadRequest>> uploads = writerContext.getUploads();
assertThat(uploads.keySet().size(), is(REPLICA_COUNT)); // Should upload to 3 replicas
@@ -178,7 +181,7 @@ public class RecordWriterTest
@Test
public void testSuccessfulWrite() throws InterruptedException
{
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
validateSuccessfulWrite(writerContext, data, COLUMN_NAMES);
}
@@ -201,7 +204,7 @@ public class RecordWriterTest
pk,
pk,
quoteIdentifiers);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
validateSuccessfulWrite(writerContext, data, columnNames);
}
@@ -209,11 +212,11 @@ public class RecordWriterTest
public void testSuccessfulWriteCheckUploads()
{
rw = new RecordWriter(writerContext, COLUMN_NAMES, () -> tc, SSTableWriter::new);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
rw.write(data);
Map<CassandraInstance, List<UploadRequest>> uploads = writerContext.getUploads();
assertThat(uploads.keySet().size(), is(REPLICA_COUNT)); // Should upload to 3 replicas
- assertThat(uploads.values().stream().mapToInt(List::size).sum(), is(REPLICA_COUNT * FILES_PER_SSTABLE * UPLOADED_TABLES));
+ assertThat(uploads.values().stream().mapToInt(List::size).sum(), is(REPLICA_COUNT * FILES_PER_SSTABLE * UPLOADED_SSTABLES));
List<UploadRequest> requests = uploads.values().stream().flatMap(List::stream).collect(Collectors.toList());
for (UploadRequest ur : requests)
{
@@ -225,7 +228,7 @@ public class RecordWriterTest
public void testWriteWithConstantTTL() throws InterruptedException
{
MockBulkWriterContext bulkWriterContext = new MockBulkWriterContext(tokenRangeMapping);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true, false, false);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true, false, false);
validateSuccessfulWrite(bulkWriterContext, data, COLUMN_NAMES);
}
@@ -233,7 +236,7 @@ public class RecordWriterTest
public void testWriteWithTTLColumn() throws InterruptedException
{
MockBulkWriterContext bulkWriterContext = new MockBulkWriterContext(tokenRangeMapping);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true, true, false);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true, true, false);
String[] columnNamesWithTtl =
{
"id", "date", "course", "marks", "ttl"
@@ -245,7 +248,7 @@ public class RecordWriterTest
public void testWriteWithConstantTimestamp() throws InterruptedException
{
MockBulkWriterContext bulkWriterContext = new MockBulkWriterContext(tokenRangeMapping);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true, false, false);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true, false, false);
validateSuccessfulWrite(bulkWriterContext, data, COLUMN_NAMES);
}
@@ -253,7 +256,7 @@ public class RecordWriterTest
public void testWriteWithTimestampColumn() throws InterruptedException
{
MockBulkWriterContext bulkWriterContext = new MockBulkWriterContext(tokenRangeMapping);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true, false, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true, false, true);
String[] columnNamesWithTimestamp =
{
"id", "date", "course", "marks", "timestamp"
@@ -265,7 +268,7 @@ public class RecordWriterTest
public void testWriteWithTimestampAndTTLColumn() throws InterruptedException
{
MockBulkWriterContext bulkWriterContext = new MockBulkWriterContext(tokenRangeMapping);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true, true, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true, true, true);
String[] columnNames =
{
"id", "date", "course", "marks", "ttl", "timestamp"
@@ -281,18 +284,19 @@ public class RecordWriterTest
when(m.job().getTokenPartitioner()).thenReturn(mtp);
// Override partition's token range to span across ranges to force a split into sub-ranges
- Range<BigInteger> overlapRange = Range.closed(BigInteger.valueOf(-9223372036854775808L), BigInteger.valueOf(200000));
+ Range<BigInteger> overlapRange = Range.openClosed(BigInteger.valueOf(-9223372036854775808L), BigInteger.valueOf(200000));
when(mtp.getTokenRange(anyInt())).thenReturn(overlapRange);
rw = new RecordWriter(m, COLUMN_NAMES, () -> tc, SSTableWriter::new);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateDataWithFakeToken(ROWS_COUNT, range);
List<StreamResult> res = rw.write(data);
assertEquals(1, res.size());
assertNotEquals(overlapRange, res.get(0).tokenRange);
- final Map<CassandraInstance, List<UploadRequest>> uploads = m.getUploads();
+ assertEquals(range, res.get(0).tokenRange);
+ Map<CassandraInstance, List<UploadRequest>> uploads = m.getUploads();
// Should upload to 3 replicas
- assertEquals(uploads.keySet().size(), REPLICA_COUNT);
- assertEquals(REPLICA_COUNT * FILES_PER_SSTABLE * UPLOADED_TABLES, uploads.values().stream().mapToInt(List::size).sum());
+ assertEquals(REPLICA_COUNT, uploads.keySet().size());
+ assertEquals(REPLICA_COUNT * FILES_PER_SSTABLE * UPLOADED_SSTABLES, uploads.values().stream().mapToInt(List::size).sum());
List<UploadRequest> requests = uploads.values().stream().flatMap(List::stream).collect(Collectors.toList());
for (UploadRequest ur : requests)
{
@@ -307,22 +311,22 @@ public class RecordWriterTest
TokenPartitioner mtp = Mockito.mock(TokenPartitioner.class);
when(m.job().getTokenPartitioner()).thenReturn(mtp);
// Override partition's token range to span across ranges to force a split into sub-ranges
- Range<BigInteger> overlapRange = Range.closed(BigInteger.valueOf(-9223372036854775808L), BigInteger.valueOf(200000));
+ Range<BigInteger> overlapRange = Range.openClosed(BigInteger.valueOf(-9223372036854775808L), BigInteger.valueOf(200000));
+ Range<BigInteger> firstSubrange = Range.openClosed(BigInteger.valueOf(-9223372036854775808L), BigInteger.valueOf(100000));
when(mtp.getTokenRange(anyInt())).thenReturn(overlapRange);
rw = new RecordWriter(m, COLUMN_NAMES, () -> tc, SSTableWriter::new);
- int numRows = 3;
- // There should be 2 SSTables since the data rows across 2 batches (0-indexed)
- int numSSTables = (int) Math.ceil((float) numRows / (writerContext.getSstableBatchSize()));
- // Generate rows with specific token values that belong to the second sub-range
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateCustomData(numRows, 100001);
+ // Generate rows that belong to the first sub-range
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateDataWithFakeToken(ROWS_COUNT, firstSubrange);
List<StreamResult> res = rw.write(data);
assertEquals(1, res.size());
assertNotEquals(overlapRange, res.get(0).tokenRange);
- final Map<CassandraInstance, List<UploadRequest>> uploads = m.getUploads();
+ assertEquals(firstSubrange, res.get(0).tokenRange);
+ Map<CassandraInstance, List<UploadRequest>> uploads = m.getUploads();
// Should upload to 3 replicas
assertEquals(REPLICA_COUNT, uploads.keySet().size());
- assertEquals(REPLICA_COUNT * FILES_PER_SSTABLE * numSSTables, uploads.values().stream().mapToInt(List::size).sum());
+ assertEquals(REPLICA_COUNT * FILES_PER_SSTABLE * UPLOADED_SSTABLES,
+ uploads.values().stream().mapToInt(List::size).sum());
List<UploadRequest> requests = uploads.values().stream().flatMap(List::stream).collect(Collectors.toList());
for (UploadRequest ur : requests)
{
@@ -334,14 +338,19 @@ public class RecordWriterTest
public void testWriteWithTokensAcrossSubRanges()
{
MockBulkWriterContext m = Mockito.spy(writerContext);
+ m.setSstableDataSizeInMB(1);
TokenPartitioner mtp = Mockito.mock(TokenPartitioner.class);
when(m.job().getTokenPartitioner()).thenReturn(mtp);
// Override partition's token range to span across ranges to force a split into sub-ranges
- Range<BigInteger> overlapRange = Range.closed(BigInteger.valueOf(-9223372036854775808L), BigInteger.valueOf(200000));
+ Range<BigInteger> overlapRange = Range.openClosed(BigInteger.valueOf(-9223372036854775808L), BigInteger.valueOf(200000L));
+ Range<BigInteger> firstSubrange = Range.openClosed(BigInteger.valueOf(-9223372036854775808L), BigInteger.valueOf(100000L));
+ Range<BigInteger> secondSubrange = Range.openClosed(BigInteger.valueOf(100000L), BigInteger.valueOf(200000L));
when(mtp.getTokenRange(anyInt())).thenReturn(overlapRange);
rw = new RecordWriter(m, COLUMN_NAMES, () -> tc, SSTableWriter::new);
- int numRows = 3;
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateCustomData(numRows, 99999);
+ int numRows = 1; // generate 1 row in each range
+ Iterator<Tuple2<DecoratedKey, Object[]>> firstRangeData = generateDataWithFakeToken(numRows, firstSubrange);
+ Iterator<Tuple2<DecoratedKey, Object[]>> secondRangeData = generateDataWithFakeToken(numRows, secondSubrange);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = Iterators.concat(firstRangeData, secondRangeData);
List<StreamResult> res = rw.write(data);
// We expect 2 streams since rows belong to different sub-ranges
assertEquals(2, res.size());
@@ -364,7 +373,7 @@ public class RecordWriterTest
public void testCorruptSSTable()
{
rw = new RecordWriter(writerContext, COLUMN_NAMES, () -> tc, (wc, path) -> new SSTableWriter(tw.setOutDir(path), path));
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
// TODO: Add better error handling with human-readable exception messages in SSTableReader::new
// That way we can assert on the exception thrown here
RuntimeException ex = assertThrows(RuntimeException.class, () -> rw.write(data));
@@ -373,11 +382,13 @@ public class RecordWriterTest
@Test
public void testWriteWithOutOfRangeTokenFails()
{
+ writerContext.setSstableWriteBatchSize(ROWS_COUNT + 100); // Write all rows in the same sstable
rw = new RecordWriter(writerContext, COLUMN_NAMES, () -> tc, (wc, path) -> new SSTableWriter(tw, folder));
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, false);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(false);
RuntimeException ex = assertThrows(RuntimeException.class, () -> rw.write(data));
- assertEquals(ex.getMessage(), "java.lang.IllegalStateException: Received Token "
- + "5765203080415074583 outside of expected range [-9223372036854775807‥100000]");
+ String expectedErr = "java.lang.IllegalStateException: Received Token " +
+ "5765203080415074583 outside the expected ranges [(-9223372036854775808‥100000]]";
+ assertEquals(expectedErr, ex.getMessage());
}
@Test
@@ -385,9 +396,9 @@ public class RecordWriterTest
{
rw = new RecordWriter(writerContext, COLUMN_NAMES, () -> tc, (wc, path) -> new SSTableWriter(tw, folder));
tw.setAddRowThrows(true);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
RuntimeException ex = assertThrows(RuntimeException.class, () -> rw.write(data));
- assertEquals(ex.getMessage(), "java.lang.RuntimeException: Failed to write because addRow throws");
+ assertEquals("java.lang.RuntimeException: Failed to write because addRow throws", ex.getMessage());
}
@Test
@@ -397,7 +408,7 @@ public class RecordWriterTest
long sixtyOneMinutesInMillis = TimeUnit.MINUTES.toMillis(61);
rw = new RecordWriter(writerContext, COLUMN_NAMES, () -> tc, (wc, path) -> new SSTableWriter(tw, folder));
writerContext.setTimeProvider(() -> System.currentTimeMillis() - sixtyOneMinutesInMillis);
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
RuntimeException ex = assertThrows(RuntimeException.class, () -> rw.write(data));
assertThat(ex.getMessage(), startsWith("Time skew between Spark and Cassandra is too large. Allowable skew is 60 minutes. Spark executor time is "));
}
@@ -410,41 +421,10 @@ public class RecordWriterTest
long remoteTime = System.currentTimeMillis() - fiftyNineMinutesInMillis;
rw = new RecordWriter(writerContext, COLUMN_NAMES, () -> tc, SSTableWriter::new);
writerContext.setTimeProvider(() -> remoteTime); // Return a very low "current time" to make sure we fail if skew is too bad
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(5, true);
+ Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(true);
rw.write(data);
}
- @DisplayName("Write 20 rows, in unbuffered mode with BATCH_SIZE of 2")
- @Test()
- void writeUnbuffered() throws InterruptedException
- {
- int numberOfRows = 20;
- int expectedTables = (int) Math.ceil(numberOfRows / writerContext.getSstableBatchSize());
- int expectedUploads = REPLICA_COUNT * FILES_PER_SSTABLE * expectedTables;
- CountDownLatch uploadsLatch = new CountDownLatch(expectedUploads);
-
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(numberOfRows, true);
- writerContext.setUploadsLatch(uploadsLatch);
- validateSuccessfulWrite(writerContext, data, COLUMN_NAMES, expectedUploads, uploadsLatch);
- }
-
- @DisplayName("Write 20 rows, in buffered mode with SSTABLE_DATA_SIZE_IN_MB of 10")
- @Test()
- void writeBuffered() throws InterruptedException
- {
- int numberOfRows = 20;
- int expectedUploads = REPLICA_COUNT * FILES_PER_SSTABLE * 1;
- CountDownLatch uploadsLatch = new CountDownLatch(expectedUploads);
-
- Iterator<Tuple2<DecoratedKey, Object[]>> data = generateData(numberOfRows, true);
- writerContext.setRowBufferMode(RowBufferMode.BUFFERED);
- writerContext.setSstableDataSizeInMB(10);
- writerContext.setUploadsLatch(uploadsLatch);
-
- // only a single data sstable file is created
- validateSuccessfulWrite(writerContext, data, COLUMN_NAMES, expectedUploads, uploadsLatch);
- }
-
private void validateSuccessfulWrite(MockBulkWriterContext writerContext,
Iterator<Tuple2<DecoratedKey, Object[]>> data,
String[] columnNames) throws InterruptedException
@@ -452,7 +432,7 @@ public class RecordWriterTest
validateSuccessfulWrite(writerContext,
data,
columnNames,
- REPLICA_COUNT * FILES_PER_SSTABLE * UPLOADED_TABLES,
+ REPLICA_COUNT * FILES_PER_SSTABLE * UPLOADED_SSTABLES,
new CountDownLatch(0));
}
@@ -476,12 +456,40 @@ public class RecordWriterTest
}
}
- private Iterator<Tuple2<DecoratedKey, Object[]>> generateData(int numValues, boolean onlyInRange)
+ private Iterator<Tuple2<DecoratedKey, Object[]>> generateData(boolean onlyInRange)
{
- return generateData(numValues, onlyInRange, false, false);
+ return generateData(onlyInRange, false, false);
}
- private Iterator<Tuple2<DecoratedKey, Object[]>> generateData(int numValues, boolean onlyInRange, boolean withTTL, boolean withTimestamp)
+ private Iterator<Tuple2<DecoratedKey, Object[]>> generateData(boolean onlyInRange, boolean withTTL, boolean withTimestamp)
+ {
+ return onlyInRange
+ ? generateData(ROWS_COUNT, range, false, withTTL, withTimestamp) // accept only tokens in range
+ : generateData(ROWS_COUNT, Range.all(), false, withTTL, withTimestamp); // accept all tokens
+ }
+
+
+ // generate data with fake tokens assigend. The fake tokens are provided by the input range.
+ // Although the data generated have fake tokens, the actual tokens computed from each tuple
+ // are still ordered ascendingly.
+ private Iterator<Tuple2<DecoratedKey, Object[]>> generateDataWithFakeToken(int numValues, Range<BigInteger> range)
+ {
+ return generateData(numValues, range, /* fake tokens */ true, /* ttl */ false, /* timestamp */ false);
+ }
+
+ /**
+ * Generate test data
+ * @param rowsCount number of rows to generate; the rows are sorted
+ * @param tokenRange token range that the generated rows belong to
+ * @param fakeTokens whether to fake tokens of each row. The acutal tokens of each row are still sorted
+ * @param withTTL wthether to add the TTL
+ * @param withTimestamp whether to add the timestamp
+ * @return iterator of the test rows
+ */
+ private Iterator<Tuple2<DecoratedKey, Object[]>> generateData(int rowsCount,
+ Range<BigInteger> tokenRange,
+ boolean fakeTokens,
+ boolean withTTL, boolean withTimestamp)
{
Stream<Tuple2<DecoratedKey, Object[]>> source = IntStream.iterate(0, integer -> integer + 1).mapToObj(index -> {
Object[] columns;
@@ -515,34 +523,32 @@ public class RecordWriterTest
}
return Tuple2.apply(tokenizer.getDecoratedKey(columns), columns);
});
- if (onlyInRange)
- {
- source = source.filter(val -> range.contains(val._1.getToken()));
- }
- Stream<Tuple2<DecoratedKey, Object[]>> limitedStream = source.limit(numValues);
- if (onlyInRange)
+
+ // filter based on the actual token if the test does not want to fake tokens
+ if (!fakeTokens)
{
- return limitedStream.sorted((o1, o2) -> o1._1.compareTo(o2._1))
- .iterator();
+ source = source.filter(val -> tokenRange.contains(val._1.getToken()));
}
- return limitedStream.iterator();
- }
- private Iterator<Tuple2<DecoratedKey, Object[]>> generateCustomData(int numValues, int start)
- {
- List<Tuple2<DecoratedKey, Object[]>> res = new ArrayList<>();
- int index = start;
- for (int i = 0; i < numValues; i++)
+ Stream<Tuple2<DecoratedKey, Object[]>> limitedStream = source.limit(rowsCount);
+
+ List<Tuple2<DecoratedKey, Object[]>> sortedData = limitedStream.sorted(Comparator.comparing(tuple -> tuple._1().getToken()))
+ .collect(Collectors.toList());
+
+ if (fakeTokens)
{
- final Object[] columns =
+ // update each tuple with fake token assigned
+ // bytebuffer keys in the list are still ordered
+ BigInteger currentToken = tokenRange.lowerEndpoint().add(BigInteger.ONE); // lower end is open; thus plus 1
+ for (int i = 0; i < sortedData.size(); i++, currentToken = currentToken.add(BigInteger.ONE))
{
- index, index, "foo" + index, index
- };
- DecoratedKey dKey = tokenizer.getDecoratedKey(columns);
- res.add(Tuple2.apply(new DecoratedKey(BigInteger.valueOf(index), dKey.getKey()), columns));
- index++;
+ Tuple2<DecoratedKey, Object[]> item = sortedData.get(i);
+ DecoratedKey key = item._1();
+ item = Tuple2.apply(new DecoratedKey(currentToken, key.getKey()), item._2());
+ sortedData.set(i, item);
+ }
}
- return res.stream().iterator();
+ return sortedData.iterator();
}
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
index a620cfb..ada44d3 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
@@ -212,13 +212,13 @@ public class RingInstanceTest
BigInteger[] tokens = getTokens(partitioner, 5);
List<RingInstance> instances = getInstances(tokens, DATACENTER_1);
CassandraInstance instance1 = instances.get(0);
- CassandraInstance instance2 = instance(tokens[0], instance1.getNodeName(), instance1.getDataCenter(), "?");
+ CassandraInstance instance2 = instance(tokens[0], instance1.nodeName(), instance1.datacenter(), "?");
ReplicaAwareFailureHandler<CassandraInstance> replicationFactor3 = ntsStrategyHandler(partitioner);
ReplicationFactor repFactor = new ReplicationFactor(ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy,
ntsOptions(new String[]{DATACENTER_1 }, new int[]{3 }));
Map<String, Set<String>> writeReplicas = instances.stream()
- .collect(Collectors.groupingBy(CassandraInstance::getDataCenter,
- Collectors.mapping(CassandraInstance::getNodeName,
+ .collect(Collectors.groupingBy(CassandraInstance::datacenter,
+ Collectors.mapping(CassandraInstance::nodeName,
Collectors.toSet())));
Multimap<RingInstance, Range<BigInteger>> tokenRanges = TokenRangeMappingUtils.setupTokenRangeMap(partitioner, repFactor, instances);
TokenRangeMapping<RingInstance> tokenRange = new TokenRangeMapping<>(partitioner,
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionConsistencyTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionConsistencyTest.java
index 46f9d54..25147ca 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionConsistencyTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionConsistencyTest.java
@@ -140,7 +140,7 @@ public class StreamSessionConsistencyTest
.sum(),
equalTo(REPLICATION_FACTOR * NUMBER_DCS * FILES_PER_SSTABLE));
List<String> instances = writerContext.getUploads().keySet().stream()
- .map(CassandraInstance::getNodeName)
+ .map(CassandraInstance::nodeName)
.collect(Collectors.toList());
assertThat(instances, containsInAnyOrder(EXPECTED_INSTANCES.toArray()));
}
@@ -162,7 +162,7 @@ public class StreamSessionConsistencyTest
int numFailures = dc1Failures.get() + dc2Failures.get();
ImmutableMap<String, AtomicInteger> dcFailures = ImmutableMap.of("DC1", dc1Failures, "DC2", dc2Failures);
boolean shouldFail = calculateFailure(consistencyLevel, dc1Failures.get(), dc2Failures.get());
- writerContext.setUploadSupplier(instance -> dcFailures.get(instance.getDataCenter()).getAndDecrement() <= 0);
+ writerContext.setUploadSupplier(instance -> dcFailures.get(instance.datacenter()).getAndDecrement() <= 0);
SSTableWriter tr = new NonValidatingTestSSTableWriter(tableWriter, folder);
tr.addRow(BigInteger.valueOf(102L), COLUMN_BIND_VALUES);
tr.close(writerContext, 1);
@@ -188,7 +188,7 @@ public class StreamSessionConsistencyTest
.sum(),
equalTo(totalFilesToUpload - filesSkipped));
List<String> instances = writerContext.getUploads().keySet().stream()
- .map(CassandraInstance::getNodeName)
+ .map(CassandraInstance::nodeName)
.collect(Collectors.toList());
assertThat(instances, containsInAnyOrder(EXPECTED_INSTANCES.toArray()));
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionTest.java
index 3a8972d..3a71640 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/StreamSessionTest.java
@@ -87,7 +87,7 @@ public class StreamSessionTest
{
List<RingInstance> replicas = ss.getReplicas();
assertNotNull(replicas);
- List<String> actualInstances = replicas.stream().map(RingInstance::getNodeName).collect(Collectors.toList());
+ List<String> actualInstances = replicas.stream().map(RingInstance::nodeName).collect(Collectors.toList());
assertThat(actualInstances, containsInAnyOrder(expectedInstances.toArray()));
}
@@ -102,7 +102,7 @@ public class StreamSessionTest
executor.assertFuturesCalled();
assertThat(executor.futures.size(), equalTo(1)); // We only scheduled one SSTable
assertThat(writerContext.getUploads().values().stream().mapToInt(Collection::size).sum(), equalTo(RF * FILES_PER_SSTABLE));
- final List<String> instances = writerContext.getUploads().keySet().stream().map(CassandraInstance::getNodeName).collect(Collectors.toList());
+ final List<String> instances = writerContext.getUploads().keySet().stream().map(CassandraInstance::nodeName).collect(Collectors.toList());
assertThat(instances, containsInAnyOrder(expectedInstances.toArray()));
}
@@ -124,7 +124,7 @@ public class StreamSessionTest
runFailedUpload();
List<String> actualInstances = writerContext.getCleanedInstances().stream()
- .map(CassandraInstance::getNodeName)
+ .map(CassandraInstance::nodeName)
.collect(Collectors.toList());
assertThat(actualInstances, containsInAnyOrder(expectedInstances.toArray()));
}
@@ -136,7 +136,7 @@ public class StreamSessionTest
runFailedUpload();
List<String> actualInstances = writerContext.getCleanedInstances().stream()
- .map(CassandraInstance::getNodeName)
+ .map(CassandraInstance::nodeName)
.collect(Collectors.toList());
assertThat(actualInstances, iterableWithSize(0));
final List<UploadRequest> uploads = writerContext.getUploads().values()
@@ -167,7 +167,7 @@ public class StreamSessionTest
});
List<String> actualInstances = writerContext.getCleanedInstances().stream()
- .map(CassandraInstance::getNodeName)
+ .map(CassandraInstance::nodeName)
.collect(Collectors.toList());
assertThat(actualInstances, containsInAnyOrder(expectedInstances.toArray()));
}
@@ -215,7 +215,7 @@ public class StreamSessionTest
ss.close(); // Force "execution" of futures
executor.assertFuturesCalled();
assertThat(writerContext.getUploads().values().stream().mapToInt(Collection::size).sum(), equalTo(RF * FILES_PER_SSTABLE));
- final List<String> instances = writerContext.getUploads().keySet().stream().map(CassandraInstance::getNodeName).collect(Collectors.toList());
+ final List<String> instances = writerContext.getUploads().keySet().stream().map(CassandraInstance::nodeName).collect(Collectors.toList());
assertThat(instances, containsInAnyOrder(expectedInstances.toArray()));
}
@@ -245,7 +245,7 @@ public class StreamSessionTest
+ " in phase UploadAndCommit.", exception.getMessage());
executor.assertFuturesCalled();
assertThat(writerContext.getUploads().values().stream().mapToInt(Collection::size).sum(), equalTo(RF * FILES_PER_SSTABLE));
- List<String> instances = writerContext.getUploads().keySet().stream().map(CassandraInstance::getNodeName).collect(Collectors.toList());
+ List<String> instances = writerContext.getUploads().keySet().stream().map(CassandraInstance::nodeName).collect(Collectors.toList());
assertThat(instances, containsInAnyOrder(expectedInstances.toArray()));
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenPartitionerTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenPartitionerTest.java
index 64909cb..26c9caf 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenPartitionerTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenPartitionerTest.java
@@ -47,45 +47,70 @@ public class TokenPartitionerTest
TokenRangeMapping<RingInstance> tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(0, ImmutableMap.of("DC1", 3), 3);
partitioner = new TokenPartitioner(tokenRangeMapping, 1, 2, 1, false);
assertEquals(4, partitioner.numPartitions());
- assertEquals(0, getPartitionForToken(new BigInteger("-9223372036854775807")));
- assertEquals(0, getPartitionForToken(0));
- assertEquals(1, getPartitionForToken(1));
- assertEquals(2, getPartitionForToken(100_001));
- assertEquals(3, getPartitionForToken(200_001));
- assertEquals(3, getPartitionForToken(new BigInteger("9223372036854775807")));
+ assertEquals(0, partitionForToken(new BigInteger("-9223372036854775807")));
+ assertEquals(0, partitionForToken(0));
+ assertEquals(1, partitionForToken(1));
+ assertEquals(2, partitionForToken(100_001));
+ assertEquals(3, partitionForToken(200_001));
+ assertEquals(3, partitionForToken(new BigInteger("9223372036854775807")));
}
@Test
public void testTwoSplits()
{
+ // There are 4 unwrapped ranges; each range is further split into 2 sub-ranges
TokenRangeMapping<RingInstance> tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(0, ImmutableMap.of("DC1", 3), 3);
partitioner = new TokenPartitioner(tokenRangeMapping, 2, 2, 1, false);
- assertEquals(9, partitioner.numPartitions());
- // Exclusive Boundary: -4611686018427387903
- assertEquals(0, getPartitionForToken(new BigInteger("-4611686018427387904")));
- assertEquals(1, getPartitionForToken(new BigInteger("-4611686018427387903")));
+ // result into the following ranges:
+ // (-9223372036854775808‥-4611686018427387904]=0,
+ // (-4611686018427387904‥0]=1,
+ // (0‥50000]=2,
+ // (50000‥100000]=3,
+ // (100000‥150000]=4,
+ // (150000‥200000]=5,
+ // (200000‥4611686018427487904]=6,
+ // (4611686018427487904‥9223372036854775807]=7
+ assertEquals(8, partitioner.numPartitions());
+
+ // Partition 0 -
+ // Test with the min token of Murmur3Partitioner. It should not exit.
+ // However, spark partitioner does not permit negative values, so it assigns the token to partition 0 artificially
+ assertEquals(0, partitionForToken(new BigInteger("-9223372036854775808")));
+ assertEquals(0, partitionForToken(new BigInteger("-9223372036854775807")));
+ // Inclusive Boundary: -4611686018427387904
+ assertEquals(0, partitionForToken(new BigInteger("-4611686018427387904")));
+
+ // Partition 1 - Exclusive Boundary: -4611686018427387904
+ assertEquals(1, partitionForToken(new BigInteger("-4611686018427387903")));
// Inclusive Boundary: 0
- assertEquals(1, getPartitionForToken(0));
- assertEquals(2, getPartitionForToken(1));
- assertEquals(2, getPartitionForToken(50));
- // Exclusive Boundary: 50000
- assertEquals(3, getPartitionForToken(51000));
- assertEquals(3, getPartitionForToken(51100));
- // Inclusive Boundary: 100000
- assertEquals(4, getPartitionForToken(100001));
- assertEquals(4, getPartitionForToken(100150));
- assertEquals(4, getPartitionForToken(150000));
- // Exclusive Boundary: 150001
- assertEquals(5, getPartitionForToken(150001));
- // Inclusive Boundary: 200000
- assertEquals(5, getPartitionForToken(200000)); // boundary
- assertEquals(6, getPartitionForToken(200001));
- assertEquals(6, getPartitionForToken(new BigInteger("4611686018427388003")));
- assertEquals(6, getPartitionForToken(new BigInteger("4611686018427487903")));
- // Exclusive Boundary: 4611686018427487904
- assertEquals(7, getPartitionForToken(new BigInteger("4611686018427487904"))); // boundary
- // Exclusive Boundary: 9223372036854775807
- assertEquals(8, getPartitionForToken(new BigInteger("9223372036854775807"))); // Single token range
+ assertEquals(1, partitionForToken(0));
+
+ // Partition 2 -
+ assertEquals(2, partitionForToken(1));
+ assertEquals(2, partitionForToken(50));
+
+ // Partition 3 -
+ assertEquals(3, partitionForToken(51000));
+ assertEquals(3, partitionForToken(51100));
+
+ // Partition 4 -
+ assertEquals(4, partitionForToken(100001));
+ assertEquals(4, partitionForToken(100150));
+ assertEquals(4, partitionForToken(150000));
+
+ // Partition 5 -
+ assertEquals(5, partitionForToken(150001));
+ assertEquals(5, partitionForToken(200000));
+
+ // Partition 6 -
+ assertEquals(6, partitionForToken(200001));
+ assertEquals(6, partitionForToken(new BigInteger("4611686018427388003")));
+ assertEquals(6, partitionForToken(new BigInteger("4611686018427487904")));
+
+ // Partition 7 - Exclusive Boundary: 4611686018427487904
+ assertEquals(7, partitionForToken(new BigInteger("4611686018427487905"))); // boundary
+ // Inclusive Boundary: 9223372036854775807
+ assertEquals(7, partitionForToken(new BigInteger("9223372036854775807")));
}
// It is possible for a keyspace to replicate to fewer than all datacenters. In these cases, the
@@ -100,12 +125,12 @@ public class TokenPartitionerTest
TokenRangeMapping<RingInstance> tokenRangeMapping = TokenRangeMappingUtils.buildTokenRangeMapping(0, ImmutableMap.of("DC1", 3, "DC2", 0), 3);
partitioner = new TokenPartitioner(tokenRangeMapping, 1, 2, 1, false);
assertEquals(4, partitioner.numPartitions());
- assertEquals(0, getPartitionForToken(new BigInteger("-9223372036854775807")));
- assertEquals(0, getPartitionForToken(0));
- assertEquals(1, getPartitionForToken(100000));
- assertEquals(2, getPartitionForToken(100001));
- assertEquals(3, getPartitionForToken(200001));
- assertEquals(3, getPartitionForToken(new BigInteger("9223372036854775807")));
+ assertEquals(0, partitionForToken(new BigInteger("-9223372036854775807")));
+ assertEquals(0, partitionForToken(0));
+ assertEquals(1, partitionForToken(100000));
+ assertEquals(2, partitionForToken(100001));
+ assertEquals(3, partitionForToken(200001));
+ assertEquals(3, partitionForToken(new BigInteger("9223372036854775807")));
}
@Test
@@ -149,12 +174,12 @@ public class TokenPartitionerTest
assertThat(partitioner.numPartitions(), greaterThanOrEqualTo(200));
}
- private int getPartitionForToken(int token)
+ private int partitionForToken(int token)
{
- return getPartitionForToken(BigInteger.valueOf(token));
+ return partitionForToken(BigInteger.valueOf(token));
}
- private int getPartitionForToken(BigInteger token)
+ private int partitionForToken(BigInteger token)
{
return partitioner.getPartition(new DecoratedKey(token, ByteBuffer.allocate(0)));
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
index 2249c62..fbcdea7 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java
@@ -35,7 +35,7 @@ import com.google.common.collect.Range;
import org.apache.cassandra.sidecar.common.data.RingEntry;
import org.apache.cassandra.sidecar.common.data.TokenRangeReplicasResponse.ReplicaMetadata;
-import org.apache.cassandra.spark.bulkwriter.token.RangeUtils;
+import org.apache.cassandra.spark.utils.RangeUtils;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.common.client.InstanceStatus;
import org.apache.cassandra.spark.data.ReplicationFactor;
@@ -60,13 +60,13 @@ public final class TokenRangeMappingUtils
{
List<RingInstance> instances = getInstances(initialToken, rfByDC, instancesPerDC);
RingInstance blockedInstance = instances.stream()
- .filter(i -> i.getIpAddress().equals(blockedInstanceIp))
+ .filter(i -> i.ipAddress().equals(blockedInstanceIp))
.findFirst()
.get();
ReplicationFactor replicationFactor = getReplicationFactor(rfByDC);
Map<String, Set<String>> writeReplicas =
- instances.stream().collect(Collectors.groupingBy(RingInstance::getDataCenter,
- Collectors.mapping(RingInstance::getNodeName,
+ instances.stream().collect(Collectors.groupingBy(RingInstance::datacenter,
+ Collectors.mapping(RingInstance::nodeName,
Collectors.toSet())));
writeReplicas.replaceAll((key, value) -> {
value.removeIf(e -> value.size() > 3);
@@ -74,12 +74,12 @@ public final class TokenRangeMappingUtils
});
List<ReplicaMetadata> replicaMetadata = instances.stream()
- .map(i -> new ReplicaMetadata(i.getRingInstance().state(),
- i.getRingInstance().status(),
- i.getNodeName(),
- i.getIpAddress(),
+ .map(i -> new ReplicaMetadata(i.ringInstance().state(),
+ i.ringInstance().status(),
+ i.nodeName(),
+ i.ipAddress(),
7012,
- i.getDataCenter()))
+ i.datacenter()))
.collect(Collectors.toList());
Multimap<RingInstance, Range<BigInteger>> tokenRanges = setupTokenRangeMap(Partitioner.Murmur3Partitioner, replicationFactor, instances);
@@ -99,7 +99,7 @@ public final class TokenRangeMappingUtils
{
List<RingInstance> instances = getInstances(initialToken, rfByDC, instancesPerDC);
RingInstance instance = instances.remove(0);
- RingEntry entry = instance.getRingInstance();
+ RingEntry entry = instance.ringInstance();
RingEntry newEntry = new RingEntry.Builder()
.datacenter(entry.datacenter())
.port(entry.port())
@@ -117,8 +117,8 @@ public final class TokenRangeMappingUtils
instances.add(0, newInstance);
ReplicationFactor replicationFactor = getReplicationFactor(rfByDC);
Map<String, Set<String>> writeReplicas =
- instances.stream().collect(Collectors.groupingBy(RingInstance::getDataCenter,
- Collectors.mapping(RingInstance::getNodeName,
+ instances.stream().collect(Collectors.groupingBy(RingInstance::datacenter,
+ Collectors.mapping(RingInstance::nodeName,
Collectors.toSet())));
writeReplicas.replaceAll((key, value) -> {
value.removeIf(e -> value.size() > 3);
@@ -126,12 +126,12 @@ public final class TokenRangeMappingUtils
});
List<ReplicaMetadata> replicaMetadata = instances.stream()
- .map(i -> new ReplicaMetadata(i.getRingInstance().state(),
- i.getRingInstance().status(),
- i.getNodeName(),
- i.getIpAddress(),
+ .map(i -> new ReplicaMetadata(i.ringInstance().state(),
+ i.ringInstance().status(),
+ i.nodeName(),
+ i.ipAddress(),
7012,
- i.getDataCenter()))
+ i.datacenter()))
.collect(Collectors.toList());
Multimap<RingInstance, Range<BigInteger>> tokenRanges = setupTokenRangeMap(Partitioner.Murmur3Partitioner, replicationFactor, instances);
@@ -156,7 +156,7 @@ public final class TokenRangeMappingUtils
if (shouldUpdateToken)
{
RingInstance instance = instances.remove(0);
- RingEntry entry = instance.getRingInstance();
+ RingEntry entry = instance.ringInstance();
RingEntry newEntry = new RingEntry.Builder()
.datacenter(entry.datacenter())
.port(entry.port())
@@ -176,8 +176,8 @@ public final class TokenRangeMappingUtils
ReplicationFactor replicationFactor = getReplicationFactor(rfByDC);
Map<String, Set<String>> writeReplicas =
- instances.stream().collect(Collectors.groupingBy(RingInstance::getDataCenter,
- Collectors.mapping(RingInstance::getNodeName,
+ instances.stream().collect(Collectors.groupingBy(RingInstance::datacenter,
+ Collectors.mapping(RingInstance::nodeName,
Collectors.toSet())));
writeReplicas.replaceAll((key, value) -> {
value.removeIf(e -> value.size() > 3);
@@ -185,12 +185,12 @@ public final class TokenRangeMappingUtils
});
List<ReplicaMetadata> replicaMetadata = instances.stream()
- .map(i -> new ReplicaMetadata(i.getRingInstance().state(),
- i.getRingInstance().status(),
- i.getNodeName(),
- i.getIpAddress(),
+ .map(i -> new ReplicaMetadata(i.ringInstance().state(),
+ i.ringInstance().status(),
+ i.nodeName(),
+ i.ipAddress(),
7012,
- i.getDataCenter()))
+ i.datacenter()))
.collect(Collectors.toList());
Multimap<RingInstance, Range<BigInteger>> tokenRanges = setupTokenRangeMap(Partitioner.Murmur3Partitioner, replicationFactor, instances);
@@ -228,7 +228,7 @@ public final class TokenRangeMappingUtils
continue;
}
List<RingInstance> dcInstances = instances.stream()
- .filter(instance -> instance.getDataCenter().matches(dataCenter))
+ .filter(instance -> instance.datacenter().matches(dataCenter))
.collect(Collectors.toList());
tokenRangeMap.putAll(RangeUtils.calculateTokenRanges(dcInstances,
replicationFactor.getOptions().get(dataCenter),
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/PartitionedDataLayerTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/PartitionedDataLayerTests.java
index 6284c26..edfd70d 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/PartitionedDataLayerTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/PartitionedDataLayerTests.java
@@ -261,7 +261,7 @@ public class PartitionedDataLayerTests extends VersionRunner
DataLayer dataLayer = new JDKSerializationTests.TestPartitionedDataLayer(bridge, 4, 32, null, ring, table);
PartitionKeyFilter filter = PartitionKeyFilter.create(ByteBuffer.wrap(RandomUtils.nextBytes(10)),
- BigInteger.valueOf(-9223372036854775808L));
+ BigInteger.valueOf(-9223372036854775807L));
SSTablesSupplier supplier = dataLayer.sstables(partitionId, null, Collections.singletonList(filter));
Set<MultipleReplicasTests.TestSSTableReader> ssTableReaders =
supplier.openAll((ssTable, isRepairPrimary) -> new MultipleReplicasTests.TestSSTableReader(ssTable));
@@ -287,7 +287,7 @@ public class PartitionedDataLayerTests extends VersionRunner
public void testFiltersInRange() throws Exception
{
Map<Integer, Range<BigInteger>> reversePartitionMap = Collections.singletonMap(
- TaskContext.getPartitionId(), Range.closed(BigInteger.ONE, BigInteger.valueOf(2L)));
+ TaskContext.getPartitionId(), Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(2L)));
TokenPartitioner mockPartitioner = mock(TokenPartitioner.class);
when(mockPartitioner.reversePartitionMap()).thenReturn(reversePartitionMap);
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java
index 13482cd..549fd5e 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/partitioner/CassandraRingTests.java
@@ -43,24 +43,31 @@ public class CassandraRingTests
{
private static Range<BigInteger> mergeRanges(Collection<Range<BigInteger>> ranges)
{
- Range<BigInteger> mergedRange = Range.closedOpen(BigInteger.ZERO, BigInteger.ZERO);
+ Range<BigInteger> mergedRange = null;
for (Range<BigInteger> range : ranges)
{
- mergedRange = mergedRange.span(range);
+ if (mergedRange == null)
+ {
+ mergedRange = range;
+ }
+ else
+ {
+ mergedRange = mergedRange.span(range);
+ }
}
return mergedRange;
}
private void validateRanges(Collection<Range<BigInteger>> ranges,
- Collection<BigInteger> validTokens,
- Collection<BigInteger> invalidTokens)
+ Collection<BigInteger> enclosedTokens,
+ Collection<BigInteger> excludedTokens)
{
RangeSet<BigInteger> rangeSet = TreeRangeSet.create();
ranges.forEach(rangeSet::add);
- validTokens.forEach(token -> assertTrue(rangeSet.contains(token), token + " should have been a valid token"));
- invalidTokens.forEach(token -> assertFalse(rangeSet.contains(token)));
+ enclosedTokens.forEach(token -> assertTrue(rangeSet.contains(token), token + " should have been a valid token"));
+ excludedTokens.forEach(token -> assertFalse(rangeSet.contains(token)));
}
@Test
@@ -84,8 +91,8 @@ public class CassandraRingTests
for (CassandraInstance instance : instances)
{
assertEquals(mergeRanges(tokenRanges.get(instance)),
- Range.closed(Partitioner.Murmur3Partitioner.minToken(),
- Partitioner.Murmur3Partitioner.maxToken()));
+ Range.openClosed(Partitioner.Murmur3Partitioner.minToken(),
+ Partitioner.Murmur3Partitioner.maxToken()));
}
}
@@ -111,10 +118,10 @@ public class CassandraRingTests
// token(0) => [201 - 0] => [201 - MAX], [MIN - 0]
validateRanges(tokenRanges.get(instances.get(0)),
Arrays.asList(BigInteger.ZERO,
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(201L)),
- Arrays.asList(BigInteger.valueOf(200L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(200L),
BigInteger.valueOf(100L),
BigInteger.valueOf(1L)));
@@ -163,21 +170,21 @@ public class CassandraRingTests
// token(0) => [101 - 0] => [101 - MAX] [MIN - 0]
validateRanges(tokenRanges.get(instances.get(0)),
Arrays.asList(BigInteger.ZERO,
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(200L),
BigInteger.valueOf(101L)),
- Arrays.asList(BigInteger.valueOf(100L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(100L),
BigInteger.valueOf(1L)));
// token(100) => [201 - 100] => [201 - MAX] [MIN - 100]
validateRanges(tokenRanges.get(instances.get(1)),
Arrays.asList(BigInteger.valueOf(0L),
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(100L),
BigInteger.valueOf(201L)),
- Arrays.asList(BigInteger.valueOf(101L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(101L),
BigInteger.valueOf(200L)));
// token(200) => [1 - 200]
@@ -233,7 +240,7 @@ public class CassandraRingTests
for (CassandraInstance instance : instances)
{
assertEquals(mergeRanges(tokenRanges.get(instance)),
- Range.closed(Partitioner.Murmur3Partitioner.minToken(),
+ Range.openClosed(Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken()));
}
}
@@ -281,10 +288,10 @@ public class CassandraRingTests
// token(0) => [201 - 0] => [201 - MAX], [MIN - 0]
validateRanges(tokenRanges.get(instances.get(0)),
Arrays.asList(BigInteger.ZERO,
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(201L)),
- Arrays.asList(BigInteger.valueOf(200L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(200L),
BigInteger.valueOf(100L),
BigInteger.valueOf(1L)));
@@ -313,10 +320,10 @@ public class CassandraRingTests
// token(1) => [202 - 1] => [202 - MAX], [MIN - 1]
validateRanges(tokenRanges.get(instances.get(3)),
Arrays.asList(BigInteger.ONE,
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(202L)),
- Arrays.asList(BigInteger.valueOf(201L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(201L),
BigInteger.valueOf(101L),
BigInteger.valueOf(2L)));
@@ -386,21 +393,21 @@ public class CassandraRingTests
// token(0) => [101 - 0] => [101 - MAX] [MIN - 0]
validateRanges(tokenRanges.get(instances.get(0)),
Arrays.asList(BigInteger.ZERO,
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(200L),
BigInteger.valueOf(101L)),
- Arrays.asList(BigInteger.valueOf(100L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(100L),
BigInteger.valueOf(1L)));
// token(100) => [201 - 100] => [201 - MAX] [MIN - 100]
validateRanges(tokenRanges.get(instances.get(1)),
Arrays.asList(BigInteger.valueOf(0L),
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(100L),
BigInteger.valueOf(201L)),
- Arrays.asList(BigInteger.valueOf(101L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(101L),
BigInteger.valueOf(200L)));
// token(200) => [1 - 200]
@@ -416,21 +423,21 @@ public class CassandraRingTests
// token(1) => [102 - 1] => [102 - MAX] [MIN - 1]
validateRanges(tokenRanges.get(instances.get(3)),
Arrays.asList(BigInteger.ONE,
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(200L),
BigInteger.valueOf(102L)),
- Arrays.asList(BigInteger.valueOf(101L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(101L),
BigInteger.valueOf(2L)));
// token(101) => [202 - 101] => [202 - MAX] [MIN - 101]
validateRanges(tokenRanges.get(instances.get(4)),
Arrays.asList(BigInteger.valueOf(1L),
- Partitioner.Murmur3Partitioner.minToken(),
Partitioner.Murmur3Partitioner.maxToken(),
BigInteger.valueOf(101L),
BigInteger.valueOf(202L)),
- Arrays.asList(BigInteger.valueOf(102L),
+ Arrays.asList(Partitioner.Murmur3Partitioner.minToken(),
+ BigInteger.valueOf(102L),
BigInteger.valueOf(201L)));
// token(201) => [2 - 201]
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/partitioner/TokenPartitionerTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/partitioner/TokenPartitionerTests.java
index b7ac2bd..d6f976f 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/partitioner/TokenPartitionerTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/partitioner/TokenPartitionerTests.java
@@ -33,6 +33,7 @@ import org.junit.jupiter.api.Test;
import org.apache.cassandra.spark.TestUtils;
import org.apache.cassandra.spark.utils.RandomUtils;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.quicktheories.QuickTheory.qt;
@@ -54,7 +55,14 @@ public class TokenPartitionerTests
private void runTest(Partitioner partitioner, int numInstances, int numCores)
{
TokenPartitioner tokenPartitioner = new TokenPartitioner(TestUtils.createRing(partitioner, numInstances), 1, numCores);
- assertTrue(tokenPartitioner.numPartitions() > 1);
+ if (numInstances == 1 && numCores == 1)
+ {
+ assertEquals(1, tokenPartitioner.numPartitions());
+ }
+ else
+ {
+ assertTrue(tokenPartitioner.numPartitions() > 1);
+ }
// Generate some random tokens and verify they only exist in a single token partition
Map<BigInteger, Integer> tokens = IntStream.range(0, NUM_TOKEN_TESTS)
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/RangeUtilsTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/RangeUtilsTest.java
index a13d8b4..e1b0d50 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/RangeUtilsTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/RangeUtilsTest.java
@@ -21,11 +21,14 @@ package org.apache.cassandra.spark.utils;
import java.math.BigInteger;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import com.google.common.collect.BoundType;
import com.google.common.collect.Multimap;
import com.google.common.collect.Range;
import org.junit.jupiter.api.Test;
@@ -37,51 +40,51 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertThrows;
-public class RangeUtilsTest
+class RangeUtilsTest
{
private static final Pattern RANGE_PATTERN = Pattern.compile("^[\\\\(|\\[](-?\\d+),(-?\\d+)[\\\\)|\\]]$");
@Test
- public void testCalculateTokenRangesTenNodesRF10()
+ void testCalculateTokenRangesTenNodesRF10()
{
assertTokenRanges(10, 10,
- new String[]{"(-9223372036854775808,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
- new String[]{"(-7378697629483820647,9223372036854775807]", "[-9223372036854775808,-7378697629483820647]"},
- new String[]{"(-5534023222112865486,9223372036854775807]", "[-9223372036854775808,-5534023222112865486]"},
- new String[]{"(-3689348814741910325,9223372036854775807]", "[-9223372036854775808,-3689348814741910325]"},
- new String[]{"(-1844674407370955164,9223372036854775807]", "[-9223372036854775808,-1844674407370955164]"},
- new String[]{"(-3,9223372036854775807]", "[-9223372036854775808,-3]"},
- new String[]{"(1844674407370955158,9223372036854775807]", "[-9223372036854775808,1844674407370955158]"},
- new String[]{"(3689348814741910319,9223372036854775807]", "[-9223372036854775808,3689348814741910319]"},
- new String[]{"(5534023222112865480,9223372036854775807]", "[-9223372036854775808,5534023222112865480]"},
- new String[]{"(7378697629483820641,9223372036854775807]", "[-9223372036854775808,7378697629483820641]"});
+ new String[]{"(-9223372036854775808,9223372036854775807]"},
+ new String[]{"(-7378697629483820647,9223372036854775807]", "(-9223372036854775808,-7378697629483820647]"},
+ new String[]{"(-5534023222112865486,9223372036854775807]", "(-9223372036854775808,-5534023222112865486]"},
+ new String[]{"(-3689348814741910325,9223372036854775807]", "(-9223372036854775808,-3689348814741910325]"},
+ new String[]{"(-1844674407370955164,9223372036854775807]", "(-9223372036854775808,-1844674407370955164]"},
+ new String[]{"(-3,9223372036854775807]", "(-9223372036854775808,-3]"},
+ new String[]{"(1844674407370955158,9223372036854775807]", "(-9223372036854775808,1844674407370955158]"},
+ new String[]{"(3689348814741910319,9223372036854775807]", "(-9223372036854775808,3689348814741910319]"},
+ new String[]{"(5534023222112865480,9223372036854775807]", "(-9223372036854775808,5534023222112865480]"},
+ new String[]{"(7378697629483820641,9223372036854775807]", "(-9223372036854775808,7378697629483820641]"});
}
@Test
- public void testCalculateTokenRangesTenNodesRF7()
+ void testCalculateTokenRangesTenNodesRF7()
{
assertTokenRanges(10, 7,
- new String[]{"(-3689348814741910325,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
- new String[]{"(-1844674407370955164,9223372036854775807]", "[-9223372036854775808,-7378697629483820647]"},
- new String[]{"(-3,9223372036854775807]", "[-9223372036854775808,-5534023222112865486]"},
- new String[]{"(1844674407370955158,9223372036854775807]", "[-9223372036854775808,-3689348814741910325]"},
- new String[]{"(3689348814741910319,9223372036854775807]", "[-9223372036854775808,-1844674407370955164]"},
- new String[]{"(5534023222112865480,9223372036854775807]", "[-9223372036854775808,-3]"},
- new String[]{"(7378697629483820641,9223372036854775807]", "[-9223372036854775808,1844674407370955158]"},
+ new String[]{"(-3689348814741910325,9223372036854775807]"},
+ new String[]{"(-1844674407370955164,9223372036854775807]", "(-9223372036854775808,-7378697629483820647]"},
+ new String[]{"(-3,9223372036854775807]", "(-9223372036854775808,-5534023222112865486]"},
+ new String[]{"(1844674407370955158,9223372036854775807]", "(-9223372036854775808,-3689348814741910325]"},
+ new String[]{"(3689348814741910319,9223372036854775807]", "(-9223372036854775808,-1844674407370955164]"},
+ new String[]{"(5534023222112865480,9223372036854775807]", "(-9223372036854775808,-3]"},
+ new String[]{"(7378697629483820641,9223372036854775807]", "(-9223372036854775808,1844674407370955158]"},
new String[]{"(-9223372036854775808,3689348814741910319]"},
new String[]{"(-7378697629483820647,5534023222112865480]"},
new String[]{"(-5534023222112865486,7378697629483820641]"});
}
@Test
- public void testCalculateTokenRangesTenNodesRF5()
+ void testCalculateTokenRangesTenNodesRF5()
{
assertTokenRanges(10, 5,
- new String[]{"(-3,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
- new String[]{"(1844674407370955158,9223372036854775807]", "[-9223372036854775808,-7378697629483820647]"},
- new String[]{"(3689348814741910319,9223372036854775807]", "[-9223372036854775808,-5534023222112865486]"},
- new String[]{"(5534023222112865480,9223372036854775807]", "[-9223372036854775808,-3689348814741910325]"},
- new String[]{"(7378697629483820641,9223372036854775807]", "[-9223372036854775808,-1844674407370955164]"},
+ new String[]{"(-3,9223372036854775807]"},
+ new String[]{"(1844674407370955158,9223372036854775807]", "(-9223372036854775808,-7378697629483820647]"},
+ new String[]{"(3689348814741910319,9223372036854775807]", "(-9223372036854775808,-5534023222112865486]"},
+ new String[]{"(5534023222112865480,9223372036854775807]", "(-9223372036854775808,-3689348814741910325]"},
+ new String[]{"(7378697629483820641,9223372036854775807]", "(-9223372036854775808,-1844674407370955164]"},
new String[]{"(-9223372036854775808,-3]"},
new String[]{"(-7378697629483820647,1844674407370955158]"},
new String[]{"(-5534023222112865486,3689348814741910319]"},
@@ -90,12 +93,12 @@ public class RangeUtilsTest
}
@Test
- public void testCalculateTokenRangesTenNodesRF3()
+ void testCalculateTokenRangesTenNodesRF3()
{
assertTokenRanges(10, 3,
- new String[]{"(3689348814741910319,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
- new String[]{"(5534023222112865480,9223372036854775807]", "[-9223372036854775808,-7378697629483820647]"},
- new String[]{"(7378697629483820641,9223372036854775807]", "[-9223372036854775808,-5534023222112865486]"},
+ new String[]{"(3689348814741910319,9223372036854775807]"},
+ new String[]{"(5534023222112865480,9223372036854775807]", "(-9223372036854775808,-7378697629483820647]"},
+ new String[]{"(7378697629483820641,9223372036854775807]", "(-9223372036854775808,-5534023222112865486]"},
new String[]{"(-9223372036854775808,-3689348814741910325]"},
new String[]{"(-7378697629483820647,-1844674407370955164]"},
new String[]{"(-5534023222112865486,-3]"},
@@ -106,10 +109,10 @@ public class RangeUtilsTest
}
@Test
- public void testCalculateTokenRangesTenNodesRF1()
+ void testCalculateTokenRangesTenNodesRF1()
{
assertTokenRanges(10, 1,
- new String[]{"(7378697629483820641,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
+ new String[]{"(7378697629483820641,9223372036854775807]"},
new String[]{"(-9223372036854775808,-7378697629483820647]"},
new String[]{"(-7378697629483820647,-5534023222112865486]"},
new String[]{"(-5534023222112865486,-3689348814741910325]"},
@@ -122,47 +125,47 @@ public class RangeUtilsTest
}
@Test
- public void testCalculateTokenRangesFourNodesRF4()
+ void testCalculateTokenRangesFourNodesRF4()
{
assertTokenRanges(4, 4,
- new String[]{"(-9223372036854775808,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
- new String[]{"(-4611686018427387904,9223372036854775807]", "[-9223372036854775808,-4611686018427387904]"},
- new String[]{"(0,9223372036854775807]", "[-9223372036854775808,0]"},
- new String[]{"(4611686018427387904,9223372036854775807]", "[-9223372036854775808,4611686018427387904]"});
+ new String[]{"(-9223372036854775808,9223372036854775807]"},
+ new String[]{"(-4611686018427387904,9223372036854775807]", "(-9223372036854775808,-4611686018427387904]"},
+ new String[]{"(0,9223372036854775807]", "(-9223372036854775808,0]"},
+ new String[]{"(4611686018427387904,9223372036854775807]", "(-9223372036854775808,4611686018427387904]"});
}
@Test
- public void testCalculateTokenRangesFourNodesRF3()
+ void testCalculateTokenRangesFourNodesRF3()
{
assertTokenRanges(4, 3,
- new String[]{"(-4611686018427387904,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
- new String[]{"(0,9223372036854775807]", "[-9223372036854775808,-4611686018427387904]"},
- new String[]{"(4611686018427387904,9223372036854775807]", "[-9223372036854775808,0]"},
+ new String[]{"(-4611686018427387904,9223372036854775807]"},
+ new String[]{"(0,9223372036854775807]", "(-9223372036854775808,-4611686018427387904]"},
+ new String[]{"(4611686018427387904,9223372036854775807]", "(-9223372036854775808,0]"},
new String[]{"(-9223372036854775808,4611686018427387904]"});
}
@Test
- public void testCalculateTokenRangesFourNodesRF2()
+ void testCalculateTokenRangesFourNodesRF2()
{
assertTokenRanges(4, 2,
- new String[]{"(0,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
- new String[]{"(4611686018427387904,9223372036854775807]", "[-9223372036854775808,-4611686018427387904]"},
+ new String[]{"(0,9223372036854775807]"},
+ new String[]{"(4611686018427387904,9223372036854775807]", "(-9223372036854775808,-4611686018427387904]"},
new String[]{"(-9223372036854775808,0]"},
new String[]{"(-4611686018427387904,4611686018427387904]"});
}
@Test
- public void testCalculateTokenRangesFourNodesRF1()
+ void testCalculateTokenRangesFourNodesRF1()
{
assertTokenRanges(4, 1,
- new String[]{"(4611686018427387904,9223372036854775807]", "[-9223372036854775808,-9223372036854775808]"},
+ new String[]{"(4611686018427387904,9223372036854775807]"},
new String[]{"(-9223372036854775808,-4611686018427387904]"},
new String[]{"(-4611686018427387904,0]"},
new String[]{"(0,4611686018427387904]"});
}
- @Test()
- public void testCalculateTokenRangesRFGreaterThanNodesFails()
+ @Test
+ void testCalculateTokenRangesRFGreaterThanNodesFails()
{
assertThrows(IllegalArgumentException.class,
() -> assertTokenRanges(2, 3,
@@ -172,18 +175,70 @@ public class RangeUtilsTest
}
@Test
- public void testCalculateTokenRangesZeroNodesSucceeds()
+ void testCalculateTokenRangesZeroNodesSucceeds()
{
assertTokenRanges(0, 3);
}
+ @Test
+ void testSplitTinyRange()
+ {
+ Range<BigInteger> range = Range.openClosed(BigInteger.ZERO, BigInteger.ONE);
+ List<Range<BigInteger>> expected = Collections.singletonList(range);
+ for (int nrSplits = 1; nrSplits < 5; nrSplits++)
+ {
+ // regardless of number of splits, the output should be a list of single range
+ assertEquals(expected, RangeUtils.split(range, nrSplits));
+ }
+ }
+
+ @Test
+ void testSplitOnlyProduceOpenClosedRanges()
+ {
+ Range<BigInteger> range = Range.openClosed(BigInteger.ZERO, BigInteger.TEN);
+ for (int nrSplit = 1; nrSplit < 15; nrSplit++) // the input range can be split for at most 10 time
+ {
+ for (Range<BigInteger> subrange : RangeUtils.split(range, nrSplit))
+ {
+ assertEquals(BoundType.OPEN, subrange.lowerBoundType());
+ assertEquals(BoundType.CLOSED, subrange.upperBoundType());
+ }
+ }
+ }
+
+ @Test
+ void testSplitYieldMoreEvenRanges()
+ {
+ Range<BigInteger> range = Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(11L));
+ int nrSplits = 4;
+ List<Range<BigInteger>> expectedResult = Arrays.asList(
+ Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(3)),
+ Range.openClosed(BigInteger.valueOf(3), BigInteger.valueOf(6)),
+ Range.openClosed(BigInteger.valueOf(6), BigInteger.valueOf(9)),
+ Range.openClosed(BigInteger.valueOf(9), BigInteger.valueOf(11))
+ );
+ assertEquals(expectedResult, RangeUtils.split(range, nrSplits));
+ }
+
+ @Test
+ void testSplitNotSatisfyNrSplits()
+ {
+ Range<BigInteger> range = Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(2));
+ int nrSplits = 5;
+ List<Range<BigInteger>> expectedResult = Arrays.asList(
+ Range.openClosed(BigInteger.ZERO, BigInteger.ONE),
+ Range.openClosed(BigInteger.ONE, BigInteger.valueOf(2))
+ );
+ assertEquals(expectedResult, RangeUtils.split(range, nrSplits));
+ }
+
private static void assertTokenRanges(int nodes, int replicationFactor, String[]... ranges)
{
assertEquals(nodes, ranges.length);
BigInteger[] tokens = getTokens(Partitioner.Murmur3Partitioner, nodes);
List<CassandraInstance> instances = getInstances(tokens);
Multimap<CassandraInstance, Range<BigInteger>> allRanges =
- RangeUtils.calculateTokenRanges(instances, replicationFactor, Partitioner.Murmur3Partitioner);
+ RangeUtils.calculateTokenRanges(instances, replicationFactor, Partitioner.Murmur3Partitioner);
for (int node = 0; node < nodes; node++)
{
assertExpectedRanges(allRanges.get(instances.get(node)), ranges[node]);
diff --git a/cassandra-analytics-core/src/test/spark2/org/apache/cassandra/spark/sparksql/PartitionKeyFilterTests.java b/cassandra-analytics-core/src/test/spark2/org/apache/cassandra/spark/sparksql/PartitionKeyFilterTests.java
index 6f939ec..4928610 100644
--- a/cassandra-analytics-core/src/test/spark2/org/apache/cassandra/spark/sparksql/PartitionKeyFilterTests.java
+++ b/cassandra-analytics-core/src/test/spark2/org/apache/cassandra/spark/sparksql/PartitionKeyFilterTests.java
@@ -99,6 +99,11 @@ public class PartitionKeyFilterTests
.collect(Collectors.toList());
for (BigInteger token : boundaryTokens)
{
+ if (token.equals(partitioner.minToken()))
+ {
+ // minToken is excluded in the ring
+ continue;
+ }
// Check boundary tokens only match 1 Spark token range
PartitionKeyFilter filter = PartitionKeyFilter.create(bridge.aInt().serialize(11), token);
assertEquals(1, tokenPartitioner.subRanges().stream()
diff --git a/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/TokenRange.java b/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/TokenRange.java
index 22dd7cc..b734b21 100644
--- a/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/TokenRange.java
+++ b/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/TokenRange.java
@@ -21,28 +21,37 @@ package org.apache.cassandra.bridge;
import java.io.Serializable;
import java.math.BigInteger;
+import java.util.Objects;
import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
/**
* This is a simple implementation of a range between two {@link BigInteger} tokens.
- * It allows us avoid dependency on Guava's {@link com.google.common.collect.Range}<{@link BigInteger}>
+ * A range is responsible for the tokens between {@code (lower, upper]}, i.e. open-closed.
+ *
+ * It allows us to avoid dependency on Guava's {@link com.google.common.collect.Range}<{@link BigInteger}>
* in the interface of Cassandre Bridge (which had brought us nothing but grief in the past).
*/
public final class TokenRange implements Serializable
{
- public static final long serialVersionUID = 42L;
+ private static final long serialVersionUID = 6367860484115802919L;
- // NOTE: Internally, ranges are always of the closed-open kind
+ // NOTE: ranges are always of the open-closed kind in Cassandra, see org.apache.cassandra.dht.Range
@NotNull
private final BigInteger lowerBound;
+ private final BigInteger firstEnclosedValue;
@NotNull
private final BigInteger upperBound;
private TokenRange(@NotNull BigInteger lowerBound, @NotNull BigInteger upperBound)
{
+ if (lowerBound.compareTo(upperBound) > 0)
+ {
+ throw new IllegalArgumentException("The lower bound cannot be greater than the upper bound to compose a TokenRange. " +
+ "lowerBound: " + lowerBound + "; upperBound: " + upperBound);
+ }
this.lowerBound = lowerBound;
+ this.firstEnclosedValue = lowerBound.add(BigInteger.ONE);
this.upperBound = upperBound;
}
@@ -55,37 +64,33 @@ public final class TokenRange implements Serializable
@NotNull
public BigInteger upperEndpoint()
{
- return upperBound.subtract(BigInteger.ONE);
+ return upperBound;
}
@NotNull
- public static TokenRange singleton(@NotNull BigInteger value)
+ public BigInteger firstEnclosedValue()
{
- return new TokenRange(value, value.add(BigInteger.ONE));
+ return firstEnclosedValue;
}
@NotNull
- public static TokenRange open(@NotNull BigInteger lower, @NotNull BigInteger upper)
+ public static TokenRange singleton(@NotNull BigInteger value)
{
- return new TokenRange(lower.add(BigInteger.ONE), upper);
+ // express the single token range in the form of open-closed range
+ return new TokenRange(value.subtract(BigInteger.ONE), value);
}
@NotNull
public static TokenRange closed(@NotNull BigInteger lower, @NotNull BigInteger upper)
{
- return new TokenRange(lower, upper.add(BigInteger.ONE));
- }
-
- @NotNull
- public static TokenRange closedOpen(@NotNull BigInteger lower, @NotNull BigInteger upper)
- {
- return new TokenRange(lower, upper);
+ // expressed [lower, upper] in the form of open-closed, (lower - 1, upper]
+ return new TokenRange(lower.subtract(BigInteger.ONE), upper);
}
@NotNull
public static TokenRange openClosed(@NotNull BigInteger lower, @NotNull BigInteger upper)
{
- return new TokenRange(lower.add(BigInteger.ONE), upper.add(BigInteger.ONE));
+ return new TokenRange(lower, upper);
}
@NotNull
@@ -96,27 +101,27 @@ public final class TokenRange implements Serializable
public boolean isEmpty()
{
- return lowerBound.compareTo(upperBound) >= 0;
+ return upperBound.compareTo(lowerBound) == 0;
}
public BigInteger size()
{
- return upperBound.max(lowerBound).subtract(lowerBound);
+ return upperBound.subtract(lowerBound);
}
public boolean contains(@NotNull BigInteger value)
{
- return lowerBound.compareTo(value) <= 0 && value.compareTo(upperBound) < 0;
+ return value.compareTo(lowerBound) > 0 && value.compareTo(upperBound) <= 0;
}
public boolean encloses(@NotNull TokenRange other)
{
- return lowerBound.compareTo(other.lowerBound) <= 0 && other.upperBound.compareTo(upperBound) <= 0;
+ return other.lowerBound.compareTo(lowerBound) >= 0 && other.upperBound.compareTo(upperBound) <= 0;
}
public boolean isConnected(@NotNull TokenRange other)
{
- return lowerBound.compareTo(other.upperBound) < 0 && other.lowerBound.compareTo(upperBound) < 0;
+ return lowerBound.compareTo(other.upperBound) < 0 && upperBound.compareTo(other.lowerBound) > 0;
}
@NotNull
@@ -131,24 +136,36 @@ public final class TokenRange implements Serializable
return new TokenRange(lowerBound.min(other.lowerBound), upperBound.max(other.upperBound));
}
+ /**
+ * Defines the same equals behavior as in the Guava's Range.
+ * For ranges {@code (0, 3)} and {@code [1, 2]}, althrough the values sets of the ranges are exactly the same,
+ * they are not considered that equals to each other
+ */
@Override
- public boolean equals(@Nullable Object other)
+ public boolean equals(Object o)
{
- return other instanceof TokenRange
- && this.lowerBound.equals(((TokenRange) other).lowerBound)
- && this.upperBound.equals(((TokenRange) other).upperBound);
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+ TokenRange that = (TokenRange) o;
+ return Objects.equals(lowerBound, that.lowerBound) && Objects.equals(upperBound, that.upperBound);
}
@Override
public int hashCode()
{
- return lowerBound.hashCode() ^ upperBound.hashCode();
+ return Objects.hash(lowerBound, upperBound);
}
@Override
@NotNull
public String toString()
{
- return lowerBound + "‥" + upperBound;
+ return "(" + lowerBound + ", " + upperBound + ']';
}
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/model/TokenOwner.java
similarity index 80%
copy from cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java
copy to cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/model/TokenOwner.java
index 0494b6f..f9119b7 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java
+++ b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/model/TokenOwner.java
@@ -17,15 +17,15 @@
* under the License.
*/
-package org.apache.cassandra.spark.common.model;
+package org.apache.cassandra.spark.data.model;
-public interface CassandraInstance
+/**
+ * Token owner owns a token
+ */
+public interface TokenOwner
{
- String getToken();
-
- String getNodeName();
-
- String getDataCenter();
-
- String getIpAddress();
+ /**
+ * @return the token it owns
+ */
+ String token();
}
diff --git a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraInstance.java b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraInstance.java
index b239029..01c51f1 100644
--- a/cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraInstance.java
+++ b/cassandra-bridge/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraInstance.java
@@ -27,9 +27,10 @@ import org.apache.commons.lang.builder.EqualsBuilder;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
+import org.apache.cassandra.spark.data.model.TokenOwner;
@SuppressWarnings("WeakerAccess")
-public class CassandraInstance implements Serializable
+public class CassandraInstance implements TokenOwner, Serializable
{
private static final long serialVersionUID = 6767636627576239773L;
private final String token;
diff --git a/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/IndexDbUtils.java b/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/IndexDbUtils.java
index e289b60..5f0f12b 100644
--- a/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/IndexDbUtils.java
+++ b/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/IndexDbUtils.java
@@ -53,7 +53,7 @@ final class IndexDbUtils
@NotNull SSTable ssTable,
@NotNull Stats stats) throws IOException
{
- long searchStartOffset = SummaryDbUtils.findIndexOffsetInSummary(indexSummary, partitioner, range.lowerEndpoint());
+ long searchStartOffset = SummaryDbUtils.findIndexOffsetInSummary(indexSummary, partitioner, range.firstEnclosedValue());
// Open the Index.db, skip to nearest offset found in Summary.db and find start & end offset for the Data.db file
return findDataDbOffset(range, partitioner, ssTable, stats, searchStartOffset);
@@ -136,7 +136,7 @@ final class IndexDbUtils
previous = ReaderUtils.readPosition(in);
ReaderUtils.skipPromotedIndex(in);
}
- assert range.lowerEndpoint().compareTo(keyToken) <= 0;
+ assert range.firstEnclosedValue().compareTo(keyToken) <= 0;
// Found first token that overlaps with Spark token range because we passed the target
// by skipping the promoted index, we use the previously-read position as start
return previous;
@@ -145,11 +145,12 @@ final class IndexDbUtils
/**
* @param keyToken key token read from Index.db
* @param range spark worker token range
- * @return true if keyToken is less than the range lower bound
+ * @return true if keyToken is not enclosed in the range and less than all values in the range
*/
static boolean isLessThan(@NotNull BigInteger keyToken, @NotNull TokenRange range)
{
- return keyToken.compareTo(range.lowerEndpoint()) < 0;
+ // TokenRange is always open at the lower end
+ return keyToken.compareTo(range.lowerEndpoint()) <= 0;
}
/**
diff --git a/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/IndexReader.java b/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/IndexReader.java
index 1c57d3a..e112b5f 100644
--- a/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/IndexReader.java
+++ b/cassandra-four-zero/src/main/java/org/apache/cassandra/spark/reader/IndexReader.java
@@ -81,14 +81,14 @@ public class IndexReader implements IIndexReader
if (!rangeFilter.overlaps(this.ssTableRange))
{
LOGGER.info("Skipping non-overlapping Index.db file rangeFilter='[{},{}]' sstableRange='[{},{}]'",
- rangeFilter.tokenRange().lowerEndpoint(), rangeFilter.tokenRange().upperEndpoint(),
- this.ssTableRange.lowerEndpoint(), this.ssTableRange.upperEndpoint());
+ rangeFilter.tokenRange().firstEnclosedValue(), rangeFilter.tokenRange().upperEndpoint(),
+ this.ssTableRange.firstEnclosedValue(), this.ssTableRange.upperEndpoint());
stats.indexFileSkipped();
return;
}
skipAhead = summary.summary().getPosition(
- SummaryDbUtils.binarySearchSummary(summary.summary(), metadata.partitioner, rangeFilter.tokenRange().lowerEndpoint())
+ SummaryDbUtils.binarySearchSummary(summary.summary(), metadata.partitioner, rangeFilter.tokenRange().firstEnclosedValue())
);
stats.indexSummaryFileRead(System.nanoTime() - now);
now = System.nanoTime();
@@ -309,7 +309,7 @@ public class IndexReader implements IIndexReader
public BigInteger firstToken()
{
- return ssTableRange != null ? ssTableRange.lowerEndpoint() : null;
+ return ssTableRange != null ? ssTableRange.firstEnclosedValue() : null;
}
public BigInteger lastToken()
diff --git a/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/IndexDbTests.java b/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/IndexDbTests.java
index efeee8d..a94e5fd 100644
--- a/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/IndexDbTests.java
+++ b/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/IndexDbTests.java
@@ -228,20 +228,14 @@ public class IndexDbTests
@Test
public void testLessThan()
{
- assertTrue(IndexDbUtils.isLessThan(BigInteger.valueOf(4L), TokenRange.open(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
assertTrue(IndexDbUtils.isLessThan(BigInteger.valueOf(4L), TokenRange.openClosed(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
assertTrue(IndexDbUtils.isLessThan(BigInteger.valueOf(4L), TokenRange.closed(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
- assertTrue(IndexDbUtils.isLessThan(BigInteger.valueOf(4L), TokenRange.closedOpen(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
- assertTrue(IndexDbUtils.isLessThan(BigInteger.valueOf(5L), TokenRange.open(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
assertTrue(IndexDbUtils.isLessThan(BigInteger.valueOf(5L), TokenRange.openClosed(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
assertFalse(IndexDbUtils.isLessThan(BigInteger.valueOf(5L), TokenRange.closed(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
- assertFalse(IndexDbUtils.isLessThan(BigInteger.valueOf(5L), TokenRange.closedOpen(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
- assertFalse(IndexDbUtils.isLessThan(BigInteger.valueOf(6L), TokenRange.open(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
assertFalse(IndexDbUtils.isLessThan(BigInteger.valueOf(6L), TokenRange.openClosed(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
assertFalse(IndexDbUtils.isLessThan(BigInteger.valueOf(6L), TokenRange.closed(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
- assertFalse(IndexDbUtils.isLessThan(BigInteger.valueOf(6L), TokenRange.closedOpen(BigInteger.valueOf(5L), BigInteger.valueOf(10L))));
}
private static BigInteger token(IPartitioner iPartitioner, int value)
diff --git a/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java b/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
index 039f3f5..ce5696d 100644
--- a/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
+++ b/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/IndexOffsetTests.java
@@ -56,24 +56,27 @@ public class IndexOffsetTests
private static final Logger LOGGER = LoggerFactory.getLogger(IndexOffsetTests.class);
private static final CassandraBridgeImplementation BRIDGE = new CassandraBridgeImplementation();
@SuppressWarnings("unchecked")
- private static final Multimap<Partitioner, TokenRange> RANGES = new ImmutableMultimap.Builder<Partitioner, TokenRange>()
- .putAll(Partitioner.RandomPartitioner, TokenRange.closedOpen(BigInteger.ZERO,
- BigInteger.ZERO),
- TokenRange.closedOpen(BigInteger.ONE,
- new BigInteger("56713727820156410577229101238628035242")),
- TokenRange.closedOpen(new BigInteger("56713727820156410577229101238628035243"),
- new BigInteger("113427455640312821154458202477256070484")),
- TokenRange.closedOpen(new BigInteger("113427455640312821154458202477256070485"),
- new BigInteger("170141183460469231731687303715884105727")))
- .putAll(Partitioner.Murmur3Partitioner, TokenRange.closed(new BigInteger("-9223372036854775808"),
- new BigInteger("-9223372036854775808")),
- TokenRange.closed(new BigInteger("-9223372036854775807"),
- new BigInteger("-3074457345618258603")),
- TokenRange.closed(new BigInteger("-3074457345618258602"),
- new BigInteger("3074457345618258602")),
- TokenRange.closed(new BigInteger("3074457345618258603"),
- new BigInteger("9223372036854775807")))
- .build();
+ private static final Multimap<Partitioner, TokenRange> RANGES =
+ new ImmutableMultimap.Builder<Partitioner, TokenRange>()
+ .putAll(Partitioner.RandomPartitioner,
+ TokenRange.openClosed(BigInteger.ZERO,
+ BigInteger.ONE),
+ TokenRange.openClosed(BigInteger.ONE,
+ new BigInteger("56713727820156410577229101238628035242")),
+ TokenRange.openClosed(new BigInteger("56713727820156410577229101238628035243"),
+ new BigInteger("113427455640312821154458202477256070484")),
+ TokenRange.openClosed(new BigInteger("113427455640312821154458202477256070485"),
+ new BigInteger("170141183460469231731687303715884105727")))
+ .putAll(Partitioner.Murmur3Partitioner,
+ TokenRange.openClosed(new BigInteger("-9223372036854775808"),
+ new BigInteger("-9223372036854775807")),
+ TokenRange.openClosed(new BigInteger("-9223372036854775807"),
+ new BigInteger("-3074457345618258603")),
+ TokenRange.openClosed(new BigInteger("-3074457345618258602"),
+ new BigInteger("3074457345618258602")),
+ TokenRange.openClosed(new BigInteger("3074457345618258603"),
+ new BigInteger("9223372036854775807")))
+ .build();
@Test
@SuppressWarnings("static-access")
diff --git a/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java b/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
index 6ebf7cf..2b87784 100644
--- a/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
+++ b/cassandra-four-zero/src/test/java/org/apache/cassandra/spark/reader/SSTableReaderTests.java
@@ -774,14 +774,14 @@ public class SSTableReaderTests
SparkRangeFilter.create(TokenRange.closed(BigInteger.valueOf(5L), BigInteger.valueOf(500L))),
Collections.emptyList());
assertTrue(range1.isPresent());
- assertEquals(BigInteger.valueOf(5L), range1.get().lowerEndpoint());
+ assertEquals(BigInteger.valueOf(5L), range1.get().firstEnclosedValue());
assertEquals(BigInteger.valueOf(500L), range1.get().upperEndpoint());
Optional<TokenRange> range2 = SSTableReader.extractRange(
SparkRangeFilter.create(TokenRange.closed(BigInteger.valueOf(-10000L), BigInteger.valueOf(29593L))),
Collections.emptyList());
assertTrue(range2.isPresent());
- assertEquals(BigInteger.valueOf(-10000L), range2.get().lowerEndpoint());
+ assertEquals(BigInteger.valueOf(-10000L), range2.get().firstEnclosedValue());
assertEquals(BigInteger.valueOf(29593L), range2.get().upperEndpoint());
assertFalse(SSTableReader.extractRange(null, Collections.emptyList()).isPresent());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org