points = row.getList("points", BytesType.instance);
+
+ return PaxosRepairHistory.fromTupleBufferList(points);
}
/**
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
index b59480e05a..cbed01885b 100644
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
@@ -52,7 +52,6 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadataRef;
-import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.WrappedRunnable;
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
index 4aec19ff3c..5fe1df70bd 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
@@ -38,7 +38,6 @@ import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionInterruptedException.java b/src/java/org/apache/cassandra/db/compaction/CompactionInterruptedException.java
index 809c10c8cf..b9174ec262 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionInterruptedException.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionInterruptedException.java
@@ -17,16 +17,14 @@
*/
package org.apache.cassandra.db.compaction;
+import org.apache.cassandra.utils.Shared;
+
+@Shared
public class CompactionInterruptedException extends RuntimeException
{
private static final long serialVersionUID = -8651427062512310398L;
- public CompactionInterruptedException(CompactionInfo info)
- {
- super("Compaction interrupted: " + info);
- }
-
- public CompactionInterruptedException(String info)
+ public CompactionInterruptedException(Object info)
{
super("Compaction interrupted: " + info);
}
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index fc4dc9c721..86ab5ffc6c 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -19,11 +19,17 @@ package org.apache.cassandra.db.compaction;
import java.util.*;
import java.util.function.LongPredicate;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Ordering;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.transform.DuplicateRowChecker;
@@ -37,8 +43,14 @@ import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.index.transactions.CompactionTransaction;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
+import org.apache.cassandra.service.paxos.PaxosRepairHistory;
+import org.apache.cassandra.service.paxos.uncommitted.PaxosRows;
import org.apache.cassandra.utils.TimeUUID;
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
+import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
+
/**
* Merge multiple iterators over the content of sstable into a "compacted" iterator.
*
@@ -110,7 +122,10 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
? EmptyIterators.unfilteredPartition(controller.cfs.metadata())
: UnfilteredPartitionIterators.merge(scanners, listener());
merged = Transformation.apply(merged, new GarbageSkipper(controller));
- merged = Transformation.apply(merged, new Purger(controller, nowInSec));
+ Transformation purger = isPaxos(controller.cfs) && paxosStatePurging() != legacy
+ ? new PaxosPurger(nowInSec)
+ : new Purger(controller, nowInSec);
+ merged = Transformation.apply(merged, purger);
merged = DuplicateRowChecker.duringCompaction(merged, type);
compacted = Transformation.apply(merged, new AbortableUnfilteredPartitionTransformation(this));
}
@@ -556,6 +571,78 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
}
+ private class PaxosPurger extends Transformation
+ {
+ private final long nowInSec;
+ private final long paxosPurgeGraceMicros = DatabaseDescriptor.getPaxosPurgeGrace(MICROSECONDS);
+ private final Map tableIdToHistory = new HashMap<>();
+ private Token currentToken;
+ private int compactedUnfiltered;
+
+ private PaxosPurger(long nowInSec)
+ {
+ this.nowInSec = nowInSec;
+ }
+
+ protected void onEmptyPartitionPostPurge(DecoratedKey key)
+ {
+ if (type == OperationType.COMPACTION)
+ controller.cfs.invalidateCachedPartition(key);
+ }
+
+ protected void updateProgress()
+ {
+ if ((++compactedUnfiltered) % UNFILTERED_TO_UPDATE_PROGRESS == 0)
+ updateBytesRead();
+ }
+
+ @Override
+ @SuppressWarnings("resource")
+ protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
+ {
+ currentToken = partition.partitionKey().getToken();
+ UnfilteredRowIterator purged = Transformation.apply(partition, this);
+ if (purged.isEmpty())
+ {
+ onEmptyPartitionPostPurge(purged.partitionKey());
+ purged.close();
+ return null;
+ }
+
+ return purged;
+ }
+
+ @Override
+ protected Row applyToRow(Row row)
+ {
+ updateProgress();
+ TableId tableId = PaxosRows.getTableId(row);
+
+ switch (paxosStatePurging())
+ {
+ default: throw new AssertionError();
+ case legacy:
+ case gc_grace:
+ {
+ TableMetadata metadata = Schema.instance.getTableMetadata(tableId);
+ return row.purgeDataOlderThan(TimeUnit.SECONDS.toMicros(nowInSec - (metadata == null ? (3 * 3600) : metadata.params.gcGraceSeconds)), false);
+ }
+ case repaired:
+ {
+ PaxosRepairHistory.Searcher history = tableIdToHistory.computeIfAbsent(tableId, find -> {
+ TableMetadata metadata = Schema.instance.getTableMetadata(find);
+ if (metadata == null)
+ return null;
+ return Keyspace.openAndGetStore(metadata).getPaxosRepairHistory().searcher();
+ });
+
+ return history == null ? row :
+ row.purgeDataOlderThan(history.ballotForToken(currentToken).unixMicros() - paxosPurgeGraceMicros, false);
+ }
+ }
+ }
+ }
+
private static class AbortableUnfilteredPartitionTransformation extends Transformation
{
private final AbortableUnfilteredRowTransformation abortableIter;
@@ -574,7 +661,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
}
- private static class AbortableUnfilteredRowTransformation extends Transformation
+ private static class AbortableUnfilteredRowTransformation extends Transformation
{
private final CompactionIterator iter;
@@ -590,4 +677,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
return row;
}
}
+
+ private static boolean isPaxos(ColumnFamilyStore cfs)
+ {
+ return cfs.name.equals(SystemKeyspace.PAXOS) && cfs.keyspace.getName().equals(SchemaConstants.SYSTEM_KEYSPACE_NAME);
+ }
}
diff --git a/src/java/org/apache/cassandra/db/marshal/ByteArrayAccessor.java b/src/java/org/apache/cassandra/db/marshal/ByteArrayAccessor.java
index bb26cba460..df24a627a4 100644
--- a/src/java/org/apache/cassandra/db/marshal/ByteArrayAccessor.java
+++ b/src/java/org/apache/cassandra/db/marshal/ByteArrayAccessor.java
@@ -29,6 +29,7 @@ import org.apache.cassandra.db.Digest;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.paxos.Ballot;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.ByteArrayUtil;
import org.apache.cassandra.utils.ByteBufferUtil;
@@ -241,6 +242,12 @@ public class ByteArrayAccessor implements ValueAccessor
return TimeUUID.fromBytes(getLong(value, 0), getLong(value, 8));
}
+ @Override
+ public Ballot toBallot(byte[] value)
+ {
+ return Ballot.deserialize(value);
+ }
+
@Override
public int putShort(byte[] dst, int offset, short value)
{
diff --git a/src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java b/src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java
index e428b7cd04..40a3bf4b34 100644
--- a/src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java
+++ b/src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java
@@ -28,6 +28,7 @@ import org.apache.cassandra.db.Digest;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.paxos.Ballot;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FastByteOperations;
@@ -245,6 +246,12 @@ public class ByteBufferAccessor implements ValueAccessor
return TimeUUID.fromBytes(value.getLong(value.position()), value.getLong(value.position() + 8));
}
+ @Override
+ public Ballot toBallot(ByteBuffer value)
+ {
+ return Ballot.deserialize(value);
+ }
+
@Override
public int putShort(ByteBuffer dst, int offset, short value)
{
diff --git a/src/java/org/apache/cassandra/db/marshal/TupleType.java b/src/java/org/apache/cassandra/db/marshal/TupleType.java
index 83fbb25d54..cc08487658 100644
--- a/src/java/org/apache/cassandra/db/marshal/TupleType.java
+++ b/src/java/org/apache/cassandra/db/marshal/TupleType.java
@@ -205,9 +205,17 @@ public class TupleType extends AbstractType
*/
public ByteBuffer[] split(ByteBuffer value)
{
- ByteBuffer[] components = new ByteBuffer[size()];
+ return split(value, size(), this);
+ }
+
+ /**
+ * Split a tuple value into its component values.
+ */
+ public static ByteBuffer[] split(ByteBuffer value, int numberOfElements, TupleType type)
+ {
+ ByteBuffer[] components = new ByteBuffer[numberOfElements];
ByteBuffer input = value.duplicate();
- for (int i = 0; i < size(); i++)
+ for (int i = 0; i < numberOfElements; i++)
{
if (!input.hasRemaining())
return Arrays.copyOfRange(components, 0, i);
@@ -226,7 +234,7 @@ public class TupleType extends AbstractType
{
throw new InvalidRequestException(String.format(
"Expected %s %s for %s column, but got more",
- size(), size() == 1 ? "value" : "values", this.asCQL3Type()));
+ numberOfElements, numberOfElements == 1 ? "value" : "values", type.asCQL3Type()));
}
return components;
diff --git a/src/java/org/apache/cassandra/db/marshal/ValueAccessor.java b/src/java/org/apache/cassandra/db/marshal/ValueAccessor.java
index 2f089a6c49..a51836e65a 100644
--- a/src/java/org/apache/cassandra/db/marshal/ValueAccessor.java
+++ b/src/java/org/apache/cassandra/db/marshal/ValueAccessor.java
@@ -37,6 +37,7 @@ import org.apache.cassandra.db.rows.CellPath;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.service.paxos.Ballot;
import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.db.ClusteringPrefix.Kind.*;
@@ -326,6 +327,9 @@ public interface ValueAccessor
/** returns a TimeUUID from offset 0 */
TimeUUID toTimeUUID(V value);
+ /** returns a TimeUUID from offset 0 */
+ Ballot toBallot(V value);
+
/**
* writes the short value {@param value} to {@param dst} at offset {@param offset}
* @return the number of bytes written to {@param value}
diff --git a/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java b/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java
index 1d6603eec2..b51a6787a0 100644
--- a/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java
+++ b/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java
@@ -375,26 +375,52 @@ public abstract class AbstractBTreePartition implements Partition, Iterable
@Override
public String toString()
{
- StringBuilder sb = new StringBuilder();
+ return toString(true);
+ }
- sb.append(String.format("[%s] key=%s partition_deletion=%s columns=%s",
- metadata(),
- metadata().partitionKeyType.getString(partitionKey().getKey()),
- partitionLevelDeletion(),
- columns()));
+ public String toString(boolean includeFullDetails)
+ {
+ StringBuilder sb = new StringBuilder();
+ if (includeFullDetails)
+ {
+ sb.append(String.format("[%s.%s] key=%s partition_deletion=%s columns=%s",
+ metadata().keyspace,
+ metadata().name,
+ metadata().partitionKeyType.getString(partitionKey().getKey()),
+ partitionLevelDeletion(),
+ columns()));
+ }
+ else
+ {
+ sb.append("key=").append(metadata().partitionKeyType.getString(partitionKey().getKey()));
+ }
if (staticRow() != Rows.EMPTY_STATIC_ROW)
- sb.append("\n ").append(staticRow().toString(metadata(), true));
+ sb.append("\n ").append(staticRow().toString(metadata(), includeFullDetails));
try (UnfilteredRowIterator iter = unfilteredIterator())
{
while (iter.hasNext())
- sb.append("\n ").append(iter.next().toString(metadata(), true));
+ sb.append("\n ").append(iter.next().toString(metadata(), includeFullDetails));
}
-
return sb.toString();
}
+ @Override
+ public boolean equals(Object obj)
+ {
+ if (!(obj instanceof PartitionUpdate))
+ return false;
+
+ PartitionUpdate that = (PartitionUpdate) obj;
+ Holder a = this.holder(), b = that.holder();
+ return partitionKey.equals(that.partitionKey)
+ && metadata().id.equals(that.metadata().id)
+ && a.deletionInfo.equals(b.deletionInfo)
+ && a.staticRow.equals(b.staticRow)
+ && Iterators.elementsEqual(iterator(), that.iterator());
+ }
+
public int rowCount()
{
return BTree.size(holder().tree);
diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
index 9530e10d3a..f6fd259e88 100644
--- a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
+++ b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
@@ -17,13 +17,16 @@
*/
package org.apache.cassandra.db.partitions;
+import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import com.google.common.collect.Iterables;
+import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
+import com.google.common.primitives.Ints;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,6 +41,9 @@ import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
+import org.apache.cassandra.utils.vint.VIntCoding;
+
+import static org.apache.cassandra.db.rows.UnfilteredRowIteratorSerializer.IS_EMPTY;
/**
* Stores updates made on a partition.
@@ -322,23 +328,19 @@ public class PartitionUpdate extends AbstractBTreePartition
*/
public int dataSize()
{
- int size = 0;
+ return Ints.saturatedCast(BTree.accumulate(holder.tree, (row, value) -> row.dataSize() + value, 0L)
+ + holder.staticRow.dataSize() + holder.deletionInfo.dataSize());
+ }
- if (holder.staticRow != null)
- {
- for (ColumnData cd : holder.staticRow.columnData())
- {
- size += cd.dataSize();
- }
- }
-
- for (Row row : this)
- {
- size += row.clustering().dataSize();
- for (ColumnData cd : row)
- size += cd.dataSize();
- }
- return size;
+ /**
+ * The size of the data contained in this update.
+ *
+ * @return the size of the data contained in this update.
+ */
+ public long unsharedHeapSize()
+ {
+ return BTree.accumulate(holder.tree, (row, value) -> row.unsharedHeapSize() + value, 0L)
+ + holder.staticRow.unsharedHeapSize() + holder.deletionInfo.unsharedHeapSize();
}
public TableMetadata metadata()
@@ -667,6 +669,21 @@ public class PartitionUpdate extends AbstractBTreePartition
false);
}
+ public static boolean isEmpty(ByteBuffer in, DeserializationHelper.Flag flag, DecoratedKey key) throws IOException
+ {
+ int position = in.position();
+ position += 16; // CFMetaData.serializer.deserialize(in, version);
+ if (position >= in.limit())
+ throw new EOFException();
+ // DecoratedKey key = metadata.decorateKey(ByteBufferUtil.readWithVIntLength(in));
+ int keyLength = (int) VIntCoding.getUnsignedVInt(in, position);
+ position += keyLength + VIntCoding.computeUnsignedVIntSize(keyLength);
+ if (position >= in.limit())
+ throw new EOFException();
+ int flags = in.get(position) & 0xff;
+ return (flags & IS_EMPTY) != 0;
+ }
+
public long serializedSize(PartitionUpdate update, int version)
{
try (UnfilteredRowIterator iter = update.unfilteredIterator())
diff --git a/src/java/org/apache/cassandra/db/rows/AbstractCell.java b/src/java/org/apache/cassandra/db/rows/AbstractCell.java
index 5d4fc3c19e..21d2c10e98 100644
--- a/src/java/org/apache/cassandra/db/rows/AbstractCell.java
+++ b/src/java/org/apache/cassandra/db/rows/AbstractCell.java
@@ -98,6 +98,11 @@ public abstract class AbstractCell extends Cell
return this;
}
+ public Cell purgeDataOlderThan(long timestamp)
+ {
+ return this.timestamp() < timestamp ? null : this;
+ }
+
public Cell copy(AbstractAllocator allocator)
{
CellPath path = path();
diff --git a/src/java/org/apache/cassandra/db/rows/BTreeRow.java b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
index 2d3ee83de5..64b494f953 100644
--- a/src/java/org/apache/cassandra/db/rows/BTreeRow.java
+++ b/src/java/org/apache/cassandra/db/rows/BTreeRow.java
@@ -474,6 +474,18 @@ public class BTreeRow extends AbstractRow
return transformAndFilter(newInfo, newDeletion, (cd) -> cd.purge(purger, nowInSec));
}
+ public Row purgeDataOlderThan(long timestamp, boolean enforceStrictLiveness)
+ {
+ LivenessInfo newInfo = primaryKeyLivenessInfo.timestamp() < timestamp ? LivenessInfo.EMPTY : primaryKeyLivenessInfo;
+ Deletion newDeletion = deletion.time().markedForDeleteAt() < timestamp ? Deletion.LIVE : deletion;
+
+ // when enforceStrictLiveness is set, a row is considered dead when it's PK liveness info is not present
+ if (enforceStrictLiveness && newDeletion.isLive() && newInfo.isEmpty())
+ return null;
+
+ return transformAndFilter(newInfo, newDeletion, cd -> cd.purgeDataOlderThan(timestamp));
+ }
+
private Row transformAndFilter(LivenessInfo info, Deletion deletion, Function function)
{
Object[] transformed = BTree.transformAndFilter(btree, function);
diff --git a/src/java/org/apache/cassandra/db/rows/BufferCell.java b/src/java/org/apache/cassandra/db/rows/BufferCell.java
index 85d28f8b46..171dfbb0d7 100644
--- a/src/java/org/apache/cassandra/db/rows/BufferCell.java
+++ b/src/java/org/apache/cassandra/db/rows/BufferCell.java
@@ -26,7 +26,6 @@ import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.db.marshal.ByteType;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
-import org.apache.cassandra.utils.memory.AbstractAllocator;
import static java.lang.String.format;
@@ -134,14 +133,6 @@ public class BufferCell extends AbstractCell
return withUpdatedValue(ByteBufferUtil.EMPTY_BYTE_BUFFER);
}
- public Cell copy(AbstractAllocator allocator)
- {
- if (!value.hasRemaining())
- return this;
-
- return new BufferCell(column, timestamp, ttl, localDeletionTime, allocator.clone(value), path == null ? null : path.copy(allocator));
- }
-
@Override
public long unsharedHeapSize()
{
diff --git a/src/java/org/apache/cassandra/db/rows/Cell.java b/src/java/org/apache/cassandra/db/rows/Cell.java
index 38d1c84880..2a19a986b3 100644
--- a/src/java/org/apache/cassandra/db/rows/Cell.java
+++ b/src/java/org/apache/cassandra/db/rows/Cell.java
@@ -167,6 +167,10 @@ public abstract class Cell extends ColumnData
// Overrides super type to provide a more precise return type.
public abstract Cell purge(DeletionPurger purger, int nowInSec);
+ @Override
+ // Overrides super type to provide a more precise return type.
+ public abstract Cell purgeDataOlderThan(long timestamp);
+
/**
* The serialization format for cell is:
* [ flags ][ timestamp ][ deletion time ][ ttl ][ path size ][ path ][ value size ][ value ]
diff --git a/src/java/org/apache/cassandra/db/rows/CellPath.java b/src/java/org/apache/cassandra/db/rows/CellPath.java
index cee24840b5..817a2d0314 100644
--- a/src/java/org/apache/cassandra/db/rows/CellPath.java
+++ b/src/java/org/apache/cassandra/db/rows/CellPath.java
@@ -65,6 +65,8 @@ public abstract class CellPath implements IMeasurableMemory
public abstract long unsharedHeapSizeExcludingData();
+ public abstract long unsharedHeapSize();
+
@Override
public final int hashCode()
{
diff --git a/src/java/org/apache/cassandra/db/rows/ColumnData.java b/src/java/org/apache/cassandra/db/rows/ColumnData.java
index 4146946616..8ac19cc496 100644
--- a/src/java/org/apache/cassandra/db/rows/ColumnData.java
+++ b/src/java/org/apache/cassandra/db/rows/ColumnData.java
@@ -58,6 +58,8 @@ public abstract class ColumnData implements IMeasurableMemory
public abstract long unsharedHeapSizeExcludingData();
+ public abstract long unsharedHeapSize();
+
/**
* Validate the column data.
*
@@ -95,6 +97,7 @@ public abstract class ColumnData implements IMeasurableMemory
public abstract ColumnData markCounterLocalToBeCleared();
public abstract ColumnData purge(DeletionPurger purger, int nowInSec);
+ public abstract ColumnData purgeDataOlderThan(long timestamp);
public abstract long maxTimestamp();
}
diff --git a/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java b/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java
index bf7714d9df..84f52e10bd 100644
--- a/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java
+++ b/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java
@@ -121,13 +121,10 @@ public class ComplexColumnData extends ColumnData implements Iterable>
return size;
}
- @Override
public long unsharedHeapSize()
{
- long heapSize = EMPTY_SIZE + ObjectSizes.sizeOfArray(cells);
- for (Cell cell : this)
- heapSize += cell.unsharedHeapSize();
- return heapSize;
+ long heapSize = EMPTY_SIZE + ObjectSizes.sizeOfArray(cells) + complexDeletion.unsharedHeapSize();
+ return BTree.| accumulate(cells, (cell, value) -> value + cell.unsharedHeapSize(), heapSize);
}
@Override
@@ -205,6 +202,12 @@ public class ComplexColumnData extends ColumnData implements Iterable>
return transformAndFilter(complexDeletion, (cell) -> filter.fetchedCellIsQueried(column, cell.path()) ? null : cell);
}
+ public ComplexColumnData purgeDataOlderThan(long timestamp)
+ {
+ DeletionTime newDeletion = complexDeletion.markedForDeleteAt() < timestamp ? DeletionTime.LIVE : complexDeletion;
+ return transformAndFilter(newDeletion, (cell) -> cell.purgeDataOlderThan(timestamp));
+ }
+
private ComplexColumnData transformAndFilter(DeletionTime newDeletion, Function, ? extends Cell> function)
{
Object[] transformed = BTree.transformAndFilter(cells, function);
diff --git a/src/java/org/apache/cassandra/db/rows/NativeCell.java b/src/java/org/apache/cassandra/db/rows/NativeCell.java
index 03dfc7090f..1267560aa2 100644
--- a/src/java/org/apache/cassandra/db/rows/NativeCell.java
+++ b/src/java/org/apache/cassandra/db/rows/NativeCell.java
@@ -177,5 +177,4 @@ public class NativeCell extends AbstractCell
{
return EMPTY_SIZE;
}
-
}
diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java
index 85d27a44eb..7575f06552 100644
--- a/src/java/org/apache/cassandra/db/rows/Row.java
+++ b/src/java/org/apache/cassandra/db/rows/Row.java
@@ -116,7 +116,7 @@ public interface Row extends Unfiltered, Iterable, IMeasurableMemory
*
* @param nowInSec the current time to decide what is deleted and what isn't
* @param enforceStrictLiveness whether the row should be purged if there is no PK liveness info,
- * normally retrieved from {@link CFMetaData#enforceStrictLiveness()}
+ * normally retrieved from {@link TableMetadata#enforceStrictLiveness()}
* @return true if there is some live information
*/
public boolean hasLiveData(int nowInSec, boolean enforceStrictLiveness);
@@ -249,6 +249,11 @@ public interface Row extends Unfiltered, Iterable, IMeasurableMemory
*/
public Row withOnlyQueriedData(ColumnFilter filter);
+ /*
+ * Returns a copy of this row without any data with a timestamp older than the one provided
+ */
+ public Row purgeDataOlderThan(long timestamp, boolean enforceStrictLiveness);
+
/**
* Returns a copy of this row where all counter cells have they "local" shard marked for clearing.
*/
@@ -281,6 +286,7 @@ public interface Row extends Unfiltered, Iterable, IMeasurableMemory
public long unsharedHeapSizeExcludingData();
public String toString(TableMetadata metadata, boolean fullDetails);
+ public long unsharedHeapSize();
/**
* Apply a function to every column in a row
diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java
index 938a3eed11..11541ee358 100644
--- a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java
+++ b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java
@@ -66,7 +66,7 @@ public class UnfilteredRowIteratorSerializer
{
protected static final Logger logger = LoggerFactory.getLogger(UnfilteredRowIteratorSerializer.class);
- private static final int IS_EMPTY = 0x01;
+ public static final int IS_EMPTY = 0x01;
private static final int IS_REVERSED = 0x02;
private static final int HAS_PARTITION_DELETION = 0x04;
private static final int HAS_STATIC_ROW = 0x08;
diff --git a/src/java/org/apache/cassandra/db/view/ViewUtils.java b/src/java/org/apache/cassandra/db/view/ViewUtils.java
index c248ddcde7..55a462c31b 100644
--- a/src/java/org/apache/cassandra/db/view/ViewUtils.java
+++ b/src/java/org/apache/cassandra/db/view/ViewUtils.java
@@ -23,7 +23,6 @@ import java.util.function.Predicate;
import com.google.common.collect.Iterables;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.EndpointsForToken;
diff --git a/src/java/org/apache/cassandra/db/virtual/StreamingVirtualTable.java b/src/java/org/apache/cassandra/db/virtual/StreamingVirtualTable.java
index 1036f18426..f9fe0ef54d 100644
--- a/src/java/org/apache/cassandra/db/virtual/StreamingVirtualTable.java
+++ b/src/java/org/apache/cassandra/db/virtual/StreamingVirtualTable.java
@@ -23,10 +23,12 @@ import java.util.UUID;
import java.util.stream.Collectors;
import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.streaming.StreamingState;
+import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
@@ -35,7 +37,7 @@ public class StreamingVirtualTable extends AbstractVirtualTable
public StreamingVirtualTable(String keyspace)
{
super(parse("CREATE TABLE streaming (" +
- " id uuid,\n" +
+ " id timeuuid,\n" +
" follower boolean,\n" +
" operation text, \n" +
" peers frozen>,\n" +
@@ -76,7 +78,7 @@ public class StreamingVirtualTable extends AbstractVirtualTable
@Override
public DataSet data(DecoratedKey partitionKey)
{
- UUID id = UUIDType.instance.compose(partitionKey.getKey());
+ TimeUUID id = TimeUUIDType.instance.compose(partitionKey.getKey());
SimpleDataSet result = new SimpleDataSet(metadata());
StreamingState state = StreamManager.instance.getStreamingState(id);
if (state != null)
diff --git a/src/java/org/apache/cassandra/db/virtual/VirtualMutation.java b/src/java/org/apache/cassandra/db/virtual/VirtualMutation.java
index 09ac4a604b..3e26032383 100644
--- a/src/java/org/apache/cassandra/db/virtual/VirtualMutation.java
+++ b/src/java/org/apache/cassandra/db/virtual/VirtualMutation.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.db.virtual;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import com.google.common.base.MoreObjects;
import com.google.common.collect.ImmutableMap;
@@ -26,6 +27,7 @@ import com.google.common.collect.ImmutableMap;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.IMutation;
+import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.schema.TableId;
@@ -104,6 +106,12 @@ public final class VirtualMutation implements IMutation
return modifications.values();
}
+ @Override
+ public Supplier hintOnFailure()
+ {
+ return null;
+ }
+
@Override
public void validateIndexedColumns()
{
diff --git a/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java b/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java
index 2856f131f1..b0f8bf90e3 100644
--- a/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java
+++ b/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java
@@ -144,7 +144,7 @@ public class Murmur3Partitioner implements IPartitioner
{
static final long serialVersionUID = -5833580143318243006L;
- final long token;
+ public final long token;
public LongToken(long token)
{
@@ -204,11 +204,16 @@ public class Murmur3Partitioner implements IPartitioner
}
@Override
- public Token increaseSlightly()
+ public LongToken increaseSlightly()
{
return new LongToken(token + 1);
}
+ public LongToken decreaseSlightly()
+ {
+ return new LongToken(token - 1);
+ }
+
/**
* Reverses murmur3 to find a possible 16 byte key that generates a given token
*/
@@ -216,7 +221,7 @@ public class Murmur3Partitioner implements IPartitioner
public static ByteBuffer keyForToken(LongToken token)
{
ByteBuffer result = ByteBuffer.allocate(16);
- long[] inv = MurmurHash.inv_hash3_x64_128(new long[] {token.token, 0L});
+ long[] inv = MurmurHash.inv_hash3_x64_128(new long[]{ token.token, 0L });
result.putLong(inv[0]).putLong(inv[1]).position(0);
return result;
}
diff --git a/src/java/org/apache/cassandra/dht/Range.java b/src/java/org/apache/cassandra/dht/Range.java
index 5b2f3d9fbf..2c468990d2 100644
--- a/src/java/org/apache/cassandra/dht/Range.java
+++ b/src/java/org/apache/cassandra/dht/Range.java
@@ -483,7 +483,7 @@ public class Range> extends AbstractBounds implemen
* Given a list of unwrapped ranges sorted by left position, return an
* equivalent list of ranges but with no overlapping ranges.
*/
- private static > List> deoverlap(List> ranges)
+ public static > List> deoverlap(List> ranges)
{
if (ranges.isEmpty())
return ranges;
diff --git a/src/java/org/apache/cassandra/exceptions/CasWriteTimeoutException.java b/src/java/org/apache/cassandra/exceptions/CasWriteTimeoutException.java
index b1347645f1..32cc014da1 100644
--- a/src/java/org/apache/cassandra/exceptions/CasWriteTimeoutException.java
+++ b/src/java/org/apache/cassandra/exceptions/CasWriteTimeoutException.java
@@ -20,13 +20,14 @@ package org.apache.cassandra.exceptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.WriteType;
+
public class CasWriteTimeoutException extends WriteTimeoutException
{
public final int contentions;
public CasWriteTimeoutException(WriteType writeType, ConsistencyLevel consistency, int received, int blockFor, int contentions)
{
- super(writeType, consistency, received, blockFor, String.format("CAS operation timed out - encountered contentions: %d", contentions));
+ super(writeType, consistency, received, blockFor, String.format("CAS operation timed out: received %d of %d required responses after %d contention retries", received, blockFor, contentions));
this.contentions = contentions;
}
}
diff --git a/src/java/org/apache/cassandra/exceptions/RequestFailureException.java b/src/java/org/apache/cassandra/exceptions/RequestFailureException.java
index c75ff9c19d..a71534720b 100644
--- a/src/java/org/apache/cassandra/exceptions/RequestFailureException.java
+++ b/src/java/org/apache/cassandra/exceptions/RequestFailureException.java
@@ -35,7 +35,7 @@ public class RequestFailureException extends RequestExecutionException
this(code, buildErrorMessage(received, failureReasonByEndpoint), consistency, received, blockFor, failureReasonByEndpoint);
}
- protected RequestFailureException(ExceptionCode code, String msg, ConsistencyLevel consistency, int received, int blockFor, Map failureReasonByEndpoint)
+ public RequestFailureException(ExceptionCode code, String msg, ConsistencyLevel consistency, int received, int blockFor, Map failureReasonByEndpoint)
{
super(code, buildErrorMessage(msg, failureReasonByEndpoint));
this.consistency = consistency;
diff --git a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java
index f205900bf8..3d3476a139 100644
--- a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java
+++ b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java
@@ -36,8 +36,9 @@ public enum RequestFailureReason
READ_TOO_MANY_TOMBSTONES (1),
TIMEOUT (2),
INCOMPATIBLE_SCHEMA (3),
- READ_SIZE (4);
-
+ READ_SIZE (4),
+ NODE_DOWN (5);
+
public static final Serializer serializer = new Serializer();
public final int code;
diff --git a/src/java/org/apache/cassandra/gms/EndpointState.java b/src/java/org/apache/cassandra/gms/EndpointState.java
index 2cc9c0debc..c60a4793bd 100644
--- a/src/java/org/apache/cassandra/gms/EndpointState.java
+++ b/src/java/org/apache/cassandra/gms/EndpointState.java
@@ -34,6 +34,7 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.CassandraVersion;
+import org.apache.cassandra.utils.NullableSerializer;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@@ -41,13 +42,12 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime;
* This abstraction represents both the HeartBeatState and the ApplicationState in an EndpointState
* instance. Any state for a given endpoint can be retrieved from this instance.
*/
-
-
public class EndpointState
{
protected static final Logger logger = LoggerFactory.getLogger(EndpointState.class);
public final static IVersionedSerializer serializer = new EndpointStateSerializer();
+ public final static IVersionedSerializer nullableSerializer = NullableSerializer.wrap(serializer);
private volatile HeartBeatState hbState;
private final AtomicReference> applicationState;
@@ -255,6 +255,20 @@ public class EndpointState
{
return "EndpointState: HeartBeatState = " + hbState + ", AppStateMap = " + applicationState.get();
}
+
+ public boolean isSupersededBy(EndpointState that)
+ {
+ int thisGeneration = this.getHeartBeatState().getGeneration();
+ int thatGeneration = that.getHeartBeatState().getGeneration();
+
+ if (thatGeneration > thisGeneration)
+ return true;
+
+ if (thisGeneration > thatGeneration)
+ return false;
+
+ return Gossiper.getMaxEndpointStateVersion(that) > Gossiper.getMaxEndpointStateVersion(this);
+ }
}
class EndpointStateSerializer implements IVersionedSerializer
diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAck2.java b/src/java/org/apache/cassandra/gms/GossipDigestAck2.java
index 0e4062bb0f..ed959a54e5 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestAck2.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestAck2.java
@@ -43,11 +43,6 @@ public class GossipDigestAck2
{
this.epStateMap = epStateMap;
}
-
- Map getEndpointStateMap()
- {
- return epStateMap;
- }
}
class GossipDigestAck2Serializer implements IVersionedSerializer
diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java
index 58c1589eca..0f0199957f 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java
@@ -44,7 +44,7 @@ public class GossipDigestAck2VerbHandler extends GossipVerbHandler remoteEpStateMap = message.payload.getEndpointStateMap();
+ Map remoteEpStateMap = message.payload.epStateMap;
/* Notify the Failure Detector */
Gossiper.instance.notifyFailureDetector(remoteEpStateMap);
Gossiper.instance.applyStateLocally(remoteEpStateMap);
diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java
index 4c68cfe8c6..7ee36a4083 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -610,7 +610,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
* @param epState
* @return
*/
- int getMaxEndpointStateVersion(EndpointState epState)
+ static int getMaxEndpointStateVersion(EndpointState epState)
{
int maxVersion = epState.getHeartBeatState().getHeartBeatVersion();
for (Map.Entry state : epState.states())
@@ -1144,6 +1144,15 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
return ImmutableSet.copyOf(endpointStateMap.keySet());
}
+ public String getForEndpoint(InetAddressAndPort ep, ApplicationState state)
+ {
+ EndpointState epState = endpointStateMap.get(ep);
+ if (epState == null)
+ return null;
+ VersionedValue value = epState.getApplicationState(state);
+ return value == null ? null : value.value;
+ }
+
public int getEndpointCount()
{
return endpointStateMap.size();
@@ -1261,7 +1270,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
return ep1.getHeartBeatState().getGeneration() - ep2.getHeartBeatState().getGeneration();
}
- void notifyFailureDetector(Map remoteEpStateMap)
+ public void notifyFailureDetector(Map remoteEpStateMap)
{
for (Entry entry : remoteEpStateMap.entrySet())
{
@@ -2143,7 +2152,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
}
}
- protected boolean isInShadowRound()
+ public boolean isInShadowRound()
{
return inShadowRound;
}
diff --git a/src/java/org/apache/cassandra/gms/VersionedValue.java b/src/java/org/apache/cassandra/gms/VersionedValue.java
index 880cb98e06..63e9487d7c 100644
--- a/src/java/org/apache/cassandra/gms/VersionedValue.java
+++ b/src/java/org/apache/cassandra/gms/VersionedValue.java
@@ -190,6 +190,14 @@ public class VersionedValue implements Comparable
Long.toString(expireTime)));
}
+ @VisibleForTesting
+ public VersionedValue left(Collection tokens, long expireTime, int generation)
+ {
+ return new VersionedValue(versionString(VersionedValue.STATUS_LEFT,
+ makeTokenString(tokens),
+ Long.toString(expireTime)), generation);
+ }
+
public VersionedValue moving(Token token)
{
return new VersionedValue(VersionedValue.STATUS_MOVING + VersionedValue.DELIMITER + partitioner.getTokenFactory().toString(token));
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
index 584a80c761..07556846b9 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
@@ -54,8 +54,8 @@ import org.apache.cassandra.dht.*;
import org.apache.cassandra.hadoop.*;
import org.apache.cassandra.utils.*;
-import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
+import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Hadoop InputFormat allowing map/reduce against Cassandra rows within one ColumnFamily.
diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java
index e9d3d3c3d3..63f619848e 100644
--- a/src/java/org/apache/cassandra/index/Index.java
+++ b/src/java/org/apache/cassandra/index/Index.java
@@ -267,7 +267,27 @@ public interface Index
public Optional getBackingTable();
/**
- * Return a task which performs a blocking flush of the index's data to persistent storage.
+ * Return a task which performs a blocking flush of the index's data corresponding to the provided
+ * base table's Memtable. This may extract any necessary data from the base table's Memtable as part of the flush.
+ *
+ * This version of the method is invoked whenever we flush the base table. If the index stores no in-memory data
+ * of its own, it is safe to only implement this method.
+ *
+ * @return task to be executed by the index manager to perform the flush.
+ */
+ public default Callable getBlockingFlushTask(Memtable baseCfs)
+ {
+ return getBlockingFlushTask();
+ }
+
+ /**
+ * Return a task which performs a blocking flush of any in-memory index data to persistent storage,
+ * independent of any flush of the base table.
+ *
+ * Note that this method is only invoked outside of normal flushes: if there is no in-memory storage
+ * for this index, and it only extracts data on flush from the base table's Memtable, then it is safe to
+ * perform no work.
+ *
* @return task to be executed by the index manager to perform the flush.
*/
public Callable getBlockingFlushTask();
diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
index d75b488f5f..c296880586 100644
--- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
+++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
@@ -38,7 +38,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ExecutorPlus;
+import org.apache.cassandra.concurrent.FutureTask;
import org.apache.cassandra.concurrent.ImmediateExecutor;
+import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.statements.schema.IndexTarget;
import org.apache.cassandra.db.*;
@@ -63,6 +65,8 @@ import org.apache.cassandra.notifications.SSTableAddedNotification;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.Indexes;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.pager.SinglePartitionPager;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.ProtocolVersion;
@@ -209,13 +213,15 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
markIndexesBuilding(ImmutableSet.of(index), true, isNewCF);
- Callable initialBuildTask = null;
+ FutureTask initialBuildTask = null;
// if the index didn't register itself, we can probably assume that no initialization needs to happen
if (indexes.containsKey(indexDef.name))
{
try
{
- initialBuildTask = index.getInitializationTask();
+ Callable call = index.getInitializationTask();
+ if (call != null)
+ initialBuildTask = new FutureTask<>(call);
}
catch (Throwable t)
{
@@ -233,16 +239,20 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
// otherwise run the initialization task asynchronously with a callback to mark it built or failed
final Promise initialization = new AsyncPromise<>();
- asyncExecutor.submit(initialBuildTask)
- .addCallback(
- success -> {
- markIndexBuilt(index, true);
- initialization.trySuccess(null);
- },
- failure -> {
- logAndMarkIndexesFailed(Collections.singleton(index), failure, true);
- initialization.tryFailure(failure);
- });
+ // we want to ensure we invoke this task asynchronously, so we want to add our callback before submission
+ // to ensure the work is not completed before we register the callback and so it gets performed by us.
+ // This is because Keyspace.open("system") can transitively attempt to open Keyspace.open("system")
+ initialBuildTask.addCallback(
+ success -> {
+ markIndexBuilt(index, true);
+ initialization.trySuccess(null);
+ },
+ failure -> {
+ logAndMarkIndexesFailed(Collections.singleton(index), failure, true);
+ initialization.tryFailure(failure);
+ }
+ );
+ asyncExecutor.execute(initialBuildTask);
return initialization;
}
@@ -667,7 +677,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
if (counter.decrementAndGet() == 0)
{
inProgressBuilds.remove(indexName);
- if (!needsFullRebuild.contains(indexName) && DatabaseDescriptor.isDaemonInitialized())
+ if (!needsFullRebuild.contains(indexName) && DatabaseDescriptor.isDaemonInitialized() && Keyspace.isInitialized())
SystemKeyspace.setIndexBuilt(baseCfs.keyspace.getName(), indexName);
}
}
@@ -798,17 +808,6 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
flushIndexesBlocking(indexes, null);
}
- /**
- * Performs a blocking flush of all custom indexes
- */
- public void flushAllNonCFSBackedIndexesBlocking()
- {
- executeAllBlocking(indexes.values()
- .stream()
- .filter(index -> !index.getBackingTable().isPresent()),
- Index::getBlockingFlushTask, null);
- }
-
/**
* Performs a blocking execution of pre-join tasks of all indexes
*/
@@ -843,6 +842,18 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
FBUtilities.waitOnFutures(wait);
}
+ /**
+ * Performs a blocking flush of all custom indexes
+ */
+ public void flushAllNonCFSBackedIndexesBlocking(Memtable baseCfsMemtable)
+ {
+ executeAllBlocking(indexes.values()
+ .stream()
+ .filter(index -> !index.getBackingTable().isPresent()),
+ index -> index.getBlockingFlushTask(baseCfsMemtable),
+ null);
+ }
+
/**
* @return all indexes which are marked as built and ready to use
*/
diff --git a/src/java/org/apache/cassandra/io/FSDiskFullWriteError.java b/src/java/org/apache/cassandra/io/FSDiskFullWriteError.java
index ebb07e2c69..abf0b6ca69 100644
--- a/src/java/org/apache/cassandra/io/FSDiskFullWriteError.java
+++ b/src/java/org/apache/cassandra/io/FSDiskFullWriteError.java
@@ -28,10 +28,4 @@ public class FSDiskFullWriteError extends FSWriteError
mutationSize,
keyspace)));
}
-
- @Override
- public String toString()
- {
- return "FSDiskFullWriteError";
- }
}
diff --git a/src/java/org/apache/cassandra/io/FSError.java b/src/java/org/apache/cassandra/io/FSError.java
index 65313947a1..4c06d9c61f 100644
--- a/src/java/org/apache/cassandra/io/FSError.java
+++ b/src/java/org/apache/cassandra/io/FSError.java
@@ -24,17 +24,30 @@ import org.apache.cassandra.io.util.File;
public abstract class FSError extends IOError
{
+ final String message;
public final String path;
public FSError(Throwable cause, File path)
{
- super(cause);
- this.path = path.toString();
+ this(null, cause, path);
}
public FSError(Throwable cause, Path path)
+ {
+ this(null, cause, path);
+ }
+
+ public FSError(String message, Throwable cause, File path)
{
super(cause);
+ this.message = message;
+ this.path = path.toString();
+ }
+
+ public FSError(String message, Throwable cause, Path path)
+ {
+ super(cause);
+ this.message = message;
this.path = path.toString();
}
@@ -53,4 +66,10 @@ public abstract class FSError extends IOError
return null;
}
+
+ @Override
+ public String toString()
+ {
+ return getClass().getSimpleName() + (message != null ? ' ' + message : "") + (path != null ? " in " + path : "");
+ }
}
diff --git a/src/java/org/apache/cassandra/io/FSNoDiskAvailableForWriteError.java b/src/java/org/apache/cassandra/io/FSNoDiskAvailableForWriteError.java
index 14dcd38f2a..415f204b11 100644
--- a/src/java/org/apache/cassandra/io/FSNoDiskAvailableForWriteError.java
+++ b/src/java/org/apache/cassandra/io/FSNoDiskAvailableForWriteError.java
@@ -30,10 +30,4 @@ public class FSNoDiskAvailableForWriteError extends FSWriteError
super(new IOException(String.format("The data directories for the %s keyspace have been marked as unwritable",
keyspace)));
}
-
- @Override
- public String toString()
- {
- return "FSNoDiskAvailableForWriteError";
- }
}
diff --git a/src/java/org/apache/cassandra/io/FSReadError.java b/src/java/org/apache/cassandra/io/FSReadError.java
index 688182a324..ac15534770 100644
--- a/src/java/org/apache/cassandra/io/FSReadError.java
+++ b/src/java/org/apache/cassandra/io/FSReadError.java
@@ -39,9 +39,23 @@ public class FSReadError extends FSError
this(cause, new File(path));
}
- @Override
- public String toString()
+ public FSReadError(String message, Throwable cause, Path path)
{
- return "FSReadError in " + path;
+ super(message, cause, path);
+ }
+
+ public FSReadError(String message, Throwable cause, File path)
+ {
+ super(message, cause, path);
+ }
+
+ public FSReadError(String message, Throwable cause, String path)
+ {
+ this(message, cause, new File(path));
+ }
+
+ public FSReadError(String message, Throwable cause)
+ {
+ this(message, cause, new File(""));
}
}
diff --git a/src/java/org/apache/cassandra/io/FSWriteError.java b/src/java/org/apache/cassandra/io/FSWriteError.java
index 481795db0e..81f00bd3d8 100644
--- a/src/java/org/apache/cassandra/io/FSWriteError.java
+++ b/src/java/org/apache/cassandra/io/FSWriteError.java
@@ -44,9 +44,23 @@ public class FSWriteError extends FSError
this(cause, new File(""));
}
- @Override
- public String toString()
+ public FSWriteError(String message, Throwable cause, Path path)
{
- return "FSWriteError in " + path;
+ super(message, cause, path);
+ }
+
+ public FSWriteError(String message, Throwable cause, File path)
+ {
+ super(message, cause, path);
+ }
+
+ public FSWriteError(String message, Throwable cause, String path)
+ {
+ this(message, cause, new File(path));
+ }
+
+ public FSWriteError(String message, Throwable cause)
+ {
+ this(message, cause, new File(""));
}
}
diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
index 6224cb7bfb..eab47c4289 100644
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@ -251,7 +251,7 @@ public class CQLSSTableWriter implements Closeable
ClientState.forInternalCalls(),
options,
insert.getTimestamp(TimeUnit.MILLISECONDS.toMicros(now), options),
- (int) TimeUnit.MILLISECONDS.toSeconds(now),
+ options.getNowInSec((int) TimeUnit.MILLISECONDS.toSeconds(now)),
insert.getTimeToLive(options),
Collections.emptyMap());
diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
index 4e3ae30d35..a95cbffeac 100644
--- a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
+++ b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
@@ -281,6 +281,11 @@ public class IndexSummaryManager implements IndexSummaryManagerMBean
@VisibleForTesting
public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
{
- ExecutorUtils.shutdownNowAndWait(timeout, unit, executor);
+ if (future != null)
+ {
+ future.cancel(false);
+ future = null;
+ }
+ ExecutorUtils.shutdownAndWait(timeout, unit, executor);
}
}
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
index 8fb5ad5f0f..c5f07bc39e 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@ -34,7 +34,6 @@ import com.google.common.primitives.Longs;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.config.CassandraRelevantProperties;
-import org.apache.cassandra.io.util.File;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java
index e25b0a9740..cff1f1c2d5 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.io.sstable.format.big;
import java.util.Collection;
-import java.util.UUID;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.schema.TableMetadata;
diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
index 0e20f285eb..e8dff32fbc 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
@@ -638,7 +638,7 @@ public class BigTableWriter extends SSTableWriter
protected Throwable doAbort(Throwable accumulate)
{
- return indexFile.abort(accumulate);
+ return summary.close(indexFile.abort(accumulate));
}
@Override
diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java b/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
index 9efcfa6ab2..ee5505f6fb 100644
--- a/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
+++ b/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
@@ -36,7 +36,6 @@ import org.apache.cassandra.io.ISerializer;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
-import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.EstimatedHistogram;
import org.apache.cassandra.utils.TimeUUID;
diff --git a/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java b/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java
index 1477c5c08d..fa8fad7d57 100644
--- a/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java
+++ b/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java
@@ -20,8 +20,6 @@ package org.apache.cassandra.io.util;
import java.nio.ByteBuffer;
import java.util.Optional;
-import org.apache.cassandra.io.util.File;
-
public class ChecksummedSequentialWriter extends SequentialWriter
{
private static final SequentialWriterOption CRC_WRITER_OPTION = SequentialWriterOption.newBuilder()
diff --git a/src/java/org/apache/cassandra/io/util/File.java b/src/java/org/apache/cassandra/io/util/File.java
index 8d4a0307a3..23113b164b 100644
--- a/src/java/org/apache/cassandra/io/util/File.java
+++ b/src/java/org/apache/cassandra/io/util/File.java
@@ -18,7 +18,9 @@
package org.apache.cassandra.io.util;
+import java.io.FileNotFoundException;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.net.URI;
import java.nio.channels.FileChannel;
import java.nio.file.*; // checkstyle: permit this import
@@ -34,6 +36,7 @@ import javax.annotation.Nullable;
import com.google.common.util.concurrent.RateLimiter;
+import net.openhft.chronicle.core.util.ThrowingFunction;
import org.apache.cassandra.io.FSWriteError;
import static org.apache.cassandra.io.util.PathUtils.filename;
@@ -164,6 +167,16 @@ public class File implements Comparable
maybeFail(delete(null, null));
}
+ /**
+ * This file will be deleted, with any failures being reported with an FSError
+ * @throws FSWriteError if cannot be deleted
+ */
+ public void deleteIfExists()
+ {
+ if (path != null)
+ PathUtils.deleteIfExists(path);
+ }
+
/**
* This file will be deleted, obeying the provided rate limiter.
* @throws FSWriteError if cannot be deleted
@@ -336,6 +349,11 @@ public class File implements Comparable
return PathUtils.createFileIfNotExists(toPathForWrite());
}
+ public boolean createDirectoriesIfNotExists()
+ {
+ return PathUtils.createDirectoriesIfNotExists(toPathForWrite());
+ }
+
/**
* Try to create a directory at this path.
* Return true if a new directory was created at this path, and false otherwise.
@@ -436,12 +454,29 @@ public class File implements Comparable
PathUtils.forEachRecursive(path, path -> forEach.accept(new File(path)));
}
+ private static ThrowingFunction nulls() { return ignore -> null; }
+ private static ThrowingFunction rethrow()
+ {
+ return fail -> {
+ if (fail == null) throw new FileNotFoundException();
+ throw fail;
+ };
+ }
+ private static ThrowingFunction unchecked()
+ {
+ return fail -> {
+ if (fail == null) fail = new FileNotFoundException();
+ throw new UncheckedIOException(fail);
+ };
+ }
+
+
/**
* @return if a directory, the names of the files within; null otherwise
*/
public String[] tryListNames()
{
- return tryListNames(path, Function.identity());
+ return tryListNames(nulls());
}
/**
@@ -449,7 +484,7 @@ public class File implements Comparable
*/
public String[] tryListNames(BiPredicate filter)
{
- return tryList(path, stream -> stream.map(PathUtils::filename).filter(filename -> filter.test(this, filename)), String[]::new);
+ return tryListNames(filter, nulls());
}
/**
@@ -457,7 +492,7 @@ public class File implements Comparable
*/
public File[] tryList()
{
- return tryList(path, Function.identity());
+ return tryList(nulls());
}
/**
@@ -465,7 +500,7 @@ public class File implements Comparable
*/
public File[] tryList(Predicate filter)
{
- return tryList(path, stream -> stream.filter(filter));
+ return tryList(filter, nulls());
}
/**
@@ -473,28 +508,148 @@ public class File implements Comparable
*/
public File[] tryList(BiPredicate filter)
{
- return tryList(path, stream -> stream.filter(file -> filter.test(this, file.name())));
+ return tryList(filter, nulls());
}
- private static String[] tryListNames(Path path, Function, Stream> toFiles)
+ /**
+ * @return if a directory, the names of the files within; null otherwise
+ */
+ public String[] listNames() throws IOException
{
- if (path == null)
- return null;
- return PathUtils.tryList(path, stream -> toFiles.apply(stream.map(File::new)).map(File::name), String[]::new);
+ return tryListNames(rethrow());
}
- private static T[] tryList(Path path, Function, Stream> transformation, IntFunction constructor)
+ /**
+ * @return if a directory, the names of the files within, filtered by the provided predicate; null otherwise
+ */
+ public String[] listNames(BiPredicate filter) throws IOException
{
- if (path == null)
- return null;
- return PathUtils.tryList(path, transformation, constructor);
+ return tryListNames(filter, rethrow());
}
- private static File[] tryList(Path path, Function, Stream> toFiles)
+ /**
+ * @return if a directory, the files within; null otherwise
+ */
+ public File[] list() throws IOException
+ {
+ return tryList(rethrow());
+ }
+
+ /**
+ * @return if a directory, the files within, filtered by the provided predicate; null otherwise
+ */
+ public File[] list(Predicate filter) throws IOException
+ {
+ return tryList(filter, rethrow());
+ }
+
+ /**
+ * @return if a directory, the files within, filtered by the provided predicate; null otherwise
+ */
+ public File[] list(BiPredicate filter) throws IOException
+ {
+ return tryList(filter, rethrow());
+ }
+
+ /**
+ * @return if a directory, the names of the files within; null otherwise
+ */
+ public String[] listNamesUnchecked() throws UncheckedIOException
+ {
+ return tryListNames(unchecked());
+ }
+
+ /**
+ * @return if a directory, the names of the files within, filtered by the provided predicate; null otherwise
+ */
+ public String[] listNamesUnchecked(BiPredicate filter) throws UncheckedIOException
+ {
+ return tryListNames(filter, unchecked());
+ }
+
+ /**
+ * @return if a directory, the files within; null otherwise
+ */
+ public File[] listUnchecked() throws UncheckedIOException
+ {
+ return tryList(unchecked());
+ }
+
+ /**
+ * @return if a directory, the files within, filtered by the provided predicate; null otherwise
+ */
+ public File[] listUnchecked(Predicate filter) throws UncheckedIOException
+ {
+ return tryList(filter, unchecked());
+ }
+
+ /**
+ * @return if a directory, the files within, filtered by the provided predicate; throw an UncheckedIO exception otherwise
+ */
+ public File[] listUnchecked(BiPredicate filter) throws UncheckedIOException
+ {
+ return tryList(filter, unchecked());
+ }
+
+ /**
+ * @return if a directory, the names of the files within; null otherwise
+ */
+ public String[] tryListNames(ThrowingFunction orElse) throws T
+ {
+ return tryListNames(path, Function.identity(), orElse);
+ }
+
+ /**
+ * @return if a directory, the names of the files within, filtered by the provided predicate; null otherwise
+ */
+ public String[] tryListNames(BiPredicate filter, ThrowingFunction orElse) throws T
+ {
+ return tryList(path, stream -> stream.map(PathUtils::filename).filter(filename -> filter.test(this, filename)), String[]::new, orElse);
+ }
+
+ /**
+ * @return if a directory, the files within; null otherwise
+ */
+ private File[] tryList(ThrowingFunction orElse) throws T
+ {
+ return tryList(path, Function.identity(), orElse);
+ }
+
+ /**
+ * @return if a directory, the files within, filtered by the provided predicate; null otherwise
+ */
+ private File[] tryList(Predicate filter, ThrowingFunction orElse) throws T
+ {
+ return tryList(path, stream -> stream.filter(filter), orElse);
+ }
+
+ /**
+ * @return if a directory, the files within, filtered by the provided predicate; null otherwise
+ */
+ private File[] tryList(BiPredicate filter, ThrowingFunction orElse) throws T
+ {
+ return tryList(path, stream -> stream.filter(file -> filter.test(this, file.name())), orElse);
+ }
+
+ private static String[] tryListNames(Path path, Function, Stream> toFiles, ThrowingFunction orElse) throws T
{
if (path == null)
- return null;
- return PathUtils.tryList(path, stream -> toFiles.apply(stream.map(File::new)), File[]::new);
+ return orElse.apply(null);
+ return PathUtils.tryList(path, stream -> toFiles.apply(stream.map(File::new)).map(File::name), String[]::new, orElse);
+ }
+
+ private static V[] tryList(Path path, Function, Stream> transformation, IntFunction constructor, ThrowingFunction orElse) throws T
+ {
+ if (path == null)
+ return orElse.apply(null);
+ return PathUtils.tryList(path, transformation, constructor, orElse);
+ }
+
+ private static File[] tryList(Path path, Function, Stream> toFiles, ThrowingFunction orElse) throws T
+ {
+ if (path == null)
+ return orElse.apply(null);
+ return PathUtils.tryList(path, stream -> toFiles.apply(stream.map(File::new)), File[]::new, orElse);
}
/**
diff --git a/src/java/org/apache/cassandra/io/util/PathUtils.java b/src/java/org/apache/cassandra/io/util/PathUtils.java
index c5a1e96bdc..2e321c7e31 100644
--- a/src/java/org/apache/cassandra/io/util/PathUtils.java
+++ b/src/java/org/apache/cassandra/io/util/PathUtils.java
@@ -34,6 +34,7 @@ import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import net.openhft.chronicle.core.util.ThrowingFunction;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.io.FSError;
import org.apache.cassandra.io.FSReadError;
@@ -132,7 +133,7 @@ public final class PathUtils
}
}
- public static T[] tryList(Path path, Function, Stream> transform, IntFunction arrayFactory)
+ public static V[] tryList(Path path, Function, Stream> transform, IntFunction arrayFactory, ThrowingFunction orElse) throws T
{
try (Stream stream = Files.list(path))
{
@@ -141,7 +142,7 @@ public final class PathUtils
}
catch (IOException e)
{
- return null;
+ return orElse.apply(e);
}
}
@@ -255,6 +256,22 @@ public final class PathUtils
}
}
+ public static void deleteIfExists(Path file)
+ {
+ try
+ {
+ Files.delete(file);
+ onDeletion.accept(file);
+ }
+ catch (IOException e)
+ {
+ if (e instanceof FileNotFoundException | e instanceof NoSuchFileException)
+ return;
+
+ throw propagateUnchecked(e, file, true);
+ }
+ }
+
public static boolean tryDelete(Path file)
{
try
@@ -375,9 +392,8 @@ public final class PathUtils
{
logger.trace("Could not move file {} to {}", from, to, e);
- // TODO: this should be an FSError (either read or write)?
- // (but for now this is maintaining legacy semantics)
- throw new RuntimeException(String.format("Failed to rename %s to %s", from, to), e);
+ // TODO: try to decide if is read or write? for now, have assumed write
+ throw propagateUnchecked(String.format("Failed to rename %s to %s", from, to), e, to, true);
}
}
@@ -733,6 +749,14 @@ public final class PathUtils
* propagate an IOException as an FSWriteError, FSReadError or UncheckedIOException
*/
public static RuntimeException propagateUnchecked(IOException ioe, Path path, boolean write)
+ {
+ return propagateUnchecked(null, ioe, path, write);
+ }
+
+ /**
+ * propagate an IOException as an FSWriteError, FSReadError or UncheckedIOException
+ */
+ public static RuntimeException propagateUnchecked(String message, IOException ioe, Path path, boolean write)
{
if (ioe instanceof FileAlreadyExistsException
|| ioe instanceof NoSuchFileException
@@ -741,10 +765,10 @@ public final class PathUtils
|| ioe instanceof java.nio.file.FileSystemLoopException
|| ioe instanceof java.nio.file.NotDirectoryException
|| ioe instanceof java.nio.file.NotLinkException)
- throw new UncheckedIOException(ioe);
+ throw new UncheckedIOException(message, ioe);
- if (write) throw new FSWriteError(ioe, path);
- else throw new FSReadError(ioe, path);
+ if (write) throw new FSWriteError(message, ioe, path);
+ else throw new FSReadError(message, ioe, path);
}
/**
diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriterOption.java b/src/java/org/apache/cassandra/io/util/SequentialWriterOption.java
index 599c654421..49093c1bd3 100644
--- a/src/java/org/apache/cassandra/io/util/SequentialWriterOption.java
+++ b/src/java/org/apache/cassandra/io/util/SequentialWriterOption.java
@@ -40,6 +40,7 @@ public class SequentialWriterOption
*
*/
public static final SequentialWriterOption DEFAULT = SequentialWriterOption.newBuilder().build();
+ public static final SequentialWriterOption FINISH_ON_CLOSE = SequentialWriterOption.newBuilder().finishOnClose(true).build();
private final int bufferSize;
private final BufferType bufferType;
diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java b/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java
index 16db4925ed..cc210622d1 100644
--- a/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java
+++ b/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java
@@ -32,9 +32,12 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
+import java.util.RandomAccess;
import java.util.Set;
+import java.util.Spliterator;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
@@ -45,6 +48,9 @@ import java.util.stream.Stream;
* A collection like class for Replica objects. Since the Replica class contains inetaddress, range, and
* transient replication status, basic contains and remove methods can be ambiguous. Replicas forces you
* to be explicit about what you're checking the container for, or removing from it.
+ *
+ * TODO: there's nothing about this collection that's unique to Replicas, and the implementation
+ * could make a useful general purpose immutable list<->set
*/
public abstract class AbstractReplicaCollection> implements ReplicaCollection
{
@@ -69,8 +75,10 @@ public abstract class AbstractReplicaCollection
+ protected final static class ReplicaList implements Iterable
{
private static final Replica[] EMPTY = new Replica[0];
Replica[] contents;
@@ -125,7 +133,7 @@ public abstract class AbstractReplicaCollection comparator)
+ public ReplicaList sorted(Comparator comparator)
{
Replica[] copy = Arrays.copyOfRange(contents, begin, begin + size);
Arrays.sort(copy, comparator);
@@ -137,6 +145,37 @@ public abstract class AbstractReplicaCollection forEach)
+ {
+ for (int i = begin, end = begin + size ; i < end ; ++i)
+ forEach.accept(contents[i]);
+ }
+
+ /** see {@link ReplicaCollection#count(Predicate)}*/
+ public int count(Predicate test)
+ {
+ int count = 0;
+ for (int i = begin, end = i + size ; i < end ; ++i)
+ if (test.test(contents[i]))
+ ++count;
+ return count;
+ }
+
+ public final boolean anyMatch(Predicate predicate)
+ {
+ for (int i = begin, end = i + size ; i < end ; ++i)
+ if (predicate.test(contents[i]))
+ return true;
+ return false;
+ }
+
+ @Override
+ public Spliterator spliterator()
+ {
+ return Arrays.spliterator(contents, begin, begin + size);
+ }
+
// we implement our own iterator, because it is trivial to do so, and in monomorphic call sites
// will compile down to almost optimal indexed for loop
@Override
@@ -163,7 +202,7 @@ public abstract class AbstractReplicaCollection Iterator transformIterator(Function function)
+ public Iterator transformIterator(Function function)
{
return new Iterator()
{
@@ -186,7 +225,7 @@ public abstract class AbstractReplicaCollection filterIterator(Predicate predicate, int limit)
+ private Iterator filterIterator(Predicate predicate, int limit)
{
return new Iterator()
{
@@ -218,6 +257,12 @@ public abstract class AbstractReplicaCollection void forEach(Function function, Consumer action)
+ {
+ for (int i = begin, end = begin + size ; i < end ; ++i)
+ action.accept(function.apply(contents[i]));
+ }
+
@VisibleForTesting
public boolean equals(Object to)
{
@@ -263,6 +308,12 @@ public abstract class AbstractReplicaCollection iterator() { return list.transformIterator(toKey); }
+
+ @Override
+ public void forEach(Consumer action)
+ {
+ list.forEach(toKey, action);
+ }
}
class EntrySet extends AbstractImmutableSet>
@@ -318,7 +369,7 @@ public abstract class AbstractReplicaCollection extends AbstractList implements RandomAccess
+ {
+ final Function view;
+ final ReplicaList list;
+
+ AsList(Function view, ReplicaList list)
+ {
+ this.view = view;
+ this.list = list;
+ }
+
+ public final T get(int index)
+ {
+ return view.apply(list.get(index));
+ }
+
+ public final int size()
+ {
+ return list.size;
+ }
+
+ @Override
+ public final void forEach(Consumer forEach)
+ {
+ list.forEach(view, forEach);
+ }
+ }
+
+
protected final ReplicaList list;
AbstractReplicaCollection(ReplicaList list)
{
@@ -398,40 +478,30 @@ public abstract class AbstractReplicaCollection List asList(Function viewTransform)
+ public final List asList(Function view)
{
- return new AbstractList()
- {
- public T get(int index)
- {
- return viewTransform.apply(list.get(index));
- }
-
- public int size()
- {
- return list.size;
- }
- };
+ return new AsList<>(view, list);
}
/** see {@link ReplicaCollection#count(Predicate)}*/
- public int count(Predicate predicate)
+ public final int count(Predicate test)
{
- int count = 0;
- for (int i = 0 ; i < list.size() ; ++i)
- if (predicate.test(list.get(i)))
- ++count;
- return count;
+ return list.count(test);
+ }
+
+ public final boolean anyMatch(Predicate test)
+ {
+ return list.anyMatch(test);
}
/** see {@link ReplicaCollection#filter(Predicate)}*/
- public final C filter(Predicate predicate)
+ public final C filter(Predicate predicate)
{
return filter(predicate, Integer.MAX_VALUE);
}
/** see {@link ReplicaCollection#filter(Predicate, int)}*/
- public final C filter(Predicate predicate, int limit)
+ public final C filter(Predicate predicate, int limit)
{
if (isEmpty())
return snapshot();
@@ -475,19 +545,19 @@ public abstract class AbstractReplicaCollection filterLazily(Predicate predicate)
+ public final Iterable filterLazily(Predicate predicate)
{
return filterLazily(predicate, Integer.MAX_VALUE);
}
/** see {@link ReplicaCollection#filterLazily(Predicate,int)}*/
- public final Iterable filterLazily(Predicate predicate, int limit)
+ public final Iterable filterLazily(Predicate predicate, int limit)
{
return () -> list.filterIterator(predicate, limit);
}
/** see {@link ReplicaCollection#sorted(Comparator)}*/
- public final C sorted(Comparator comparator)
+ public final C sorted(Comparator comparator)
{
return snapshot(list.sorted(comparator));
}
@@ -512,6 +582,11 @@ public abstract class AbstractReplicaCollection forEach)
+ {
+ list.forEach(forEach);
+ }
+
public final Stream stream() { return list.stream(); }
/**
diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
index 0b7c378232..7f33c578ae 100644
--- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
+++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
@@ -24,6 +24,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
import com.google.common.base.Preconditions;
@@ -32,6 +33,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.WriteType;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.RingPosition;
@@ -128,17 +130,20 @@ public abstract class AbstractReplicationStrategy
*/
public abstract EndpointsForRange calculateNaturalReplicas(Token searchToken, TokenMetadata tokenMetadata);
- public AbstractWriteResponseHandler getWriteResponseHandler(ReplicaPlan.ForTokenWrite replicaPlan,
+ public AbstractWriteResponseHandler getWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan,
Runnable callback,
WriteType writeType,
+ Supplier hintOnFailure,
long queryStartNanoTime)
{
- return getWriteResponseHandler(replicaPlan, callback, writeType, queryStartNanoTime, DatabaseDescriptor.getIdealConsistencyLevel());
+ return getWriteResponseHandler(replicaPlan, callback, writeType, hintOnFailure,
+ queryStartNanoTime, DatabaseDescriptor.getIdealConsistencyLevel());
}
- public AbstractWriteResponseHandler getWriteResponseHandler(ReplicaPlan.ForTokenWrite replicaPlan,
+ public AbstractWriteResponseHandler getWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan,
Runnable callback,
WriteType writeType,
+ Supplier hintOnFailure,
long queryStartNanoTime,
ConsistencyLevel idealConsistencyLevel)
{
@@ -146,15 +151,15 @@ public abstract class AbstractReplicationStrategy
if (replicaPlan.consistencyLevel().isDatacenterLocal())
{
// block for in this context will be localnodes block.
- resultResponseHandler = new DatacenterWriteResponseHandler(replicaPlan, callback, writeType, queryStartNanoTime);
+ resultResponseHandler = new DatacenterWriteResponseHandler(replicaPlan, callback, writeType, hintOnFailure, queryStartNanoTime);
}
else if (replicaPlan.consistencyLevel() == ConsistencyLevel.EACH_QUORUM && (this instanceof NetworkTopologyStrategy))
{
- resultResponseHandler = new DatacenterSyncWriteResponseHandler(replicaPlan, callback, writeType, queryStartNanoTime);
+ resultResponseHandler = new DatacenterSyncWriteResponseHandler(replicaPlan, callback, writeType, hintOnFailure, queryStartNanoTime);
}
else
{
- resultResponseHandler = new WriteResponseHandler(replicaPlan, callback, writeType, queryStartNanoTime);
+ resultResponseHandler = new WriteResponseHandler(replicaPlan, callback, writeType, hintOnFailure, queryStartNanoTime);
}
//Check if tracking the ideal consistency level is configured
@@ -173,6 +178,7 @@ public abstract class AbstractReplicationStrategy
AbstractWriteResponseHandler idealHandler = getWriteResponseHandler(replicaPlan.withConsistencyLevel(idealConsistencyLevel),
callback,
writeType,
+ hintOnFailure,
queryStartNanoTime,
idealConsistencyLevel);
resultResponseHandler.setIdealCLResponseHandler(idealHandler);
diff --git a/src/java/org/apache/cassandra/locator/Endpoints.java b/src/java/org/apache/cassandra/locator/Endpoints.java
index 32e972a445..1561db2bce 100644
--- a/src/java/org/apache/cassandra/locator/Endpoints.java
+++ b/src/java/org/apache/cassandra/locator/Endpoints.java
@@ -21,7 +21,6 @@ package org.apache.cassandra.locator;
import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import org.apache.cassandra.utils.FBUtilities;
-import java.util.AbstractList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -54,20 +53,14 @@ public abstract class Endpoints> extends AbstractReplicaC
return byEndpoint().keySet();
}
+ public InetAddressAndPort endpoint(int i)
+ {
+ return get(i).endpoint();
+ }
+
public List endpointList()
{
- return new AbstractList()
- {
- public InetAddressAndPort get(int index)
- {
- return list.get(index).endpoint();
- }
-
- public int size()
- {
- return list.size;
- }
- };
+ return asList(Replica::endpoint);
}
public Map byEndpoint()
@@ -87,6 +80,11 @@ public abstract class Endpoints> extends AbstractReplicaC
replica);
}
+ public boolean contains(InetAddressAndPort endpoint)
+ {
+ return endpoint != null && byEndpoint().containsKey(endpoint);
+ }
+
public E withoutSelf()
{
InetAddressAndPort self = FBUtilities.getBroadcastAddressAndPort();
diff --git a/src/java/org/apache/cassandra/locator/EndpointsForToken.java b/src/java/org/apache/cassandra/locator/EndpointsForToken.java
index c709988762..70cd76325c 100644
--- a/src/java/org/apache/cassandra/locator/EndpointsForToken.java
+++ b/src/java/org/apache/cassandra/locator/EndpointsForToken.java
@@ -19,7 +19,11 @@
package org.apache.cassandra.locator;
import com.google.common.base.Preconditions;
+
+import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.StorageService;
import java.util.Arrays;
import java.util.Collection;
@@ -67,6 +71,11 @@ public class EndpointsForToken extends Endpoints
return new EndpointsForToken(token, newList, byEndpoint);
}
+ public Replica lookup(InetAddressAndPort endpoint)
+ {
+ return byEndpoint().get(endpoint);
+ }
+
public static class Builder extends EndpointsForToken implements ReplicaCollection.Builder
{
boolean built;
@@ -146,4 +155,34 @@ public class EndpointsForToken extends Endpoints
if (replicas.isEmpty()) return empty(token);
return builder(token, replicas.size()).addAll(replicas).build();
}
+
+ public static EndpointsForToken natural(Keyspace keyspace, Token token)
+ {
+ return keyspace.getReplicationStrategy().getNaturalReplicasForToken(token);
+ }
+
+ public static EndpointsForToken natural(AbstractReplicationStrategy replicationStrategy, Token token)
+ {
+ return replicationStrategy.getNaturalReplicasForToken(token);
+ }
+
+ public static EndpointsForToken natural(TableMetadata table, Token token)
+ {
+ return natural(Keyspace.open(table.keyspace), token);
+ }
+
+ public static EndpointsForToken pending(TableMetadata table, Token token)
+ {
+ return pending(table.keyspace, token);
+ }
+
+ public static EndpointsForToken pending(Keyspace keyspace, Token token)
+ {
+ return pending(keyspace.getName(), token);
+ }
+
+ public static EndpointsForToken pending(String keyspace, Token token)
+ {
+ return StorageService.instance.getTokenMetadata().pendingEndpointsForToken(token, keyspace);
+ }
}
diff --git a/src/java/org/apache/cassandra/locator/InOurDcTester.java b/src/java/org/apache/cassandra/locator/InOurDc.java
similarity index 71%
rename from src/java/org/apache/cassandra/locator/InOurDcTester.java
rename to src/java/org/apache/cassandra/locator/InOurDc.java
index 181caceea8..34e8ef89c3 100644
--- a/src/java/org/apache/cassandra/locator/InOurDcTester.java
+++ b/src/java/org/apache/cassandra/locator/InOurDc.java
@@ -18,11 +18,12 @@
package org.apache.cassandra.locator;
-import org.apache.cassandra.config.DatabaseDescriptor;
-
import java.util.function.Predicate;
-public class InOurDcTester
+import static org.apache.cassandra.config.DatabaseDescriptor.getEndpointSnitch;
+import static org.apache.cassandra.config.DatabaseDescriptor.getLocalDataCenter;
+
+public class InOurDc
{
private static ReplicaTester replicas;
private static EndpointTester endpoints;
@@ -30,7 +31,7 @@ public class InOurDcTester
final String dc;
final IEndpointSnitch snitch;
- private InOurDcTester(String dc, IEndpointSnitch snitch)
+ private InOurDc(String dc, IEndpointSnitch snitch)
{
this.dc = dc;
this.snitch = snitch;
@@ -38,15 +39,15 @@ public class InOurDcTester
boolean stale()
{
- return dc != DatabaseDescriptor.getLocalDataCenter()
- || snitch != DatabaseDescriptor.getEndpointSnitch()
+ return dc != getLocalDataCenter()
+ || snitch != getEndpointSnitch()
// this final clause checks if somehow the snitch/localDc have got out of whack;
// presently, this is possible but very unlikely, but this check will also help
// resolve races on these global fields as well
|| !dc.equals(snitch.getLocalDatacenter());
}
- private static final class ReplicaTester extends InOurDcTester implements Predicate
+ private static final class ReplicaTester extends InOurDc implements Predicate
{
private ReplicaTester(String dc, IEndpointSnitch snitch)
{
@@ -60,7 +61,7 @@ public class InOurDcTester
}
}
- private static final class EndpointTester extends InOurDcTester implements Predicate
+ private static final class EndpointTester extends InOurDc implements Predicate
{
private EndpointTester(String dc, IEndpointSnitch snitch)
{
@@ -78,7 +79,7 @@ public class InOurDcTester
{
ReplicaTester cur = replicas;
if (cur == null || cur.stale())
- replicas = cur = new ReplicaTester(DatabaseDescriptor.getLocalDataCenter(), DatabaseDescriptor.getEndpointSnitch());
+ replicas = cur = new ReplicaTester(getLocalDataCenter(), getEndpointSnitch());
return cur;
}
@@ -86,8 +87,18 @@ public class InOurDcTester
{
EndpointTester cur = endpoints;
if (cur == null || cur.stale())
- endpoints = cur = new EndpointTester(DatabaseDescriptor.getLocalDataCenter(), DatabaseDescriptor.getEndpointSnitch());
+ endpoints = cur = new EndpointTester(getLocalDataCenter(), getEndpointSnitch());
return cur;
}
+ public static boolean isInOurDc(Replica replica)
+ {
+ return replicas().test(replica);
+ }
+
+ public static boolean isInOurDc(InetAddressAndPort endpoint)
+ {
+ return endpoints().test(endpoint);
+ }
+
}
diff --git a/src/java/org/apache/cassandra/locator/ReplicaCollection.java b/src/java/org/apache/cassandra/locator/ReplicaCollection.java
index ec671d5aa5..b679b506b0 100644
--- a/src/java/org/apache/cassandra/locator/ReplicaCollection.java
+++ b/src/java/org/apache/cassandra/locator/ReplicaCollection.java
@@ -62,14 +62,14 @@ public interface ReplicaCollection> extends Itera
/**
* @return the number of replicas that match the predicate
*/
- public abstract int count(Predicate predicate);
+ public abstract int count(Predicate predicate);
/**
* @return a *eagerly constructed* copy of this collection containing the Replica that match the provided predicate.
* An effort will be made to either return ourself, or a subList, where possible.
* It is guaranteed that no changes to any upstream Builder will affect the state of the result.
*/
- public abstract C filter(Predicate predicate);
+ public abstract C filter(Predicate predicate);
/**
* @return a *eagerly constructed* copy of this collection containing the Replica that match the provided predicate.
@@ -77,18 +77,18 @@ public interface ReplicaCollection> extends Itera
* It is guaranteed that no changes to any upstream Builder will affect the state of the result.
* Only the first maxSize items will be returned.
*/
- public abstract C filter(Predicate predicate, int maxSize);
+ public abstract C filter(Predicate predicate, int maxSize);
/**
* @return a *lazily constructed* Iterable over this collection, containing the Replica that match the provided predicate.
*/
- public abstract Iterable filterLazily(Predicate predicate);
+ public abstract Iterable filterLazily(Predicate predicate);
/**
* @return a *lazily constructed* Iterable over this collection, containing the Replica that match the provided predicate.
* Only the first maxSize matching items will be returned.
*/
- public abstract Iterable filterLazily(Predicate predicate, int maxSize);
+ public abstract Iterable |