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

Remove Journal record ownership tags functionality 

patch by Aleksey Yeschenko; reviewed by Benedict Elliott Smith for
CASSANDRA-20229
This commit is contained in:
Aleksey Yeschenko 2025-01-15 15:37:40 +00:00 committed by David Capwell
parent 3c7cb1e3b3
commit b8a3706ef9
13 changed files with 97 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
src/java/org/apache/cassandra/journal/ActiveSegment.java
View File

@ -22,12 +22,12 @@ import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
import java.util.*;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.locks.LockSupport;
import com.codahale.metrics.Timer;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.OpOrder;
@ -362,9 +362,9 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
*/
@SuppressWarnings({ "resource", "RedundantSuppression" }) // op group will be closed by Allocation#write()
Allocation allocate(int entrySize, Set<Integer> hosts)
Allocation allocate(int entrySize)
{
int totalSize = totalEntrySize(hosts, entrySize);
int totalSize = totalEntrySize(entrySize);
OpOrder.Group opGroup = appendOrder.start();
try
{
@ -383,11 +383,12 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
}
}
private int totalEntrySize(Set<Integer> hosts, int recordSize)
private int totalEntrySize(int recordSize)
{
return EntrySerializer.fixedEntrySize(keySupport, descriptor.userVersion)
+ EntrySerializer.variableEntrySize(hosts.size())
+ recordSize;
return EntrySerializer.headerSize(keySupport, descriptor.userVersion)
+ recordSize
+ TypeSizes.INT_SIZE // CRC
;
}
// allocate bytes in the segment, or return -1 if not enough space
@ -437,12 +438,12 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
this.length = length;
}
void write(K id, ByteBuffer record, Set<Integer> hosts)
void write(K id, ByteBuffer record)
{
try
{
EntrySerializer.write(id, record, hosts, keySupport, buffer, descriptor.userVersion);
metadata.update(hosts);
EntrySerializer.write(id, record, keySupport, buffer, descriptor.userVersion);
metadata.update();
index.update(id, start, length);
}
catch (IOException e)
@ -456,13 +457,13 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
}
// Variant of write that does not allocate/return a record pointer
void writeInternal(K id, ByteBuffer record, Set<Integer> hosts)
void writeInternal(K id, ByteBuffer record)
{
try
{
EntrySerializer.write(id, record, hosts, keySupport, buffer, descriptor.userVersion);
EntrySerializer.write(id, record, keySupport, buffer, descriptor.userVersion);
index.update(id, start, length);
metadata.update(hosts);
metadata.update();
}
catch (IOException e)
{

64
src/java/org/apache/cassandra/journal/EntrySerializer.java
View File

@ -20,11 +20,9 @@ package org.apache.cassandra.journal;
import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Set;
import java.util.zip.CRC32;
import accord.utils.Invariants;
import org.agrona.collections.IntHashSet;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Crc;
@ -35,7 +33,6 @@ public final class EntrySerializer
{
static <K> void write(K key,
ByteBuffer record,
Set<Integer> hosts,
KeySupport<K> keySupport,
ByteBuffer out,
int userVersion)
@ -43,17 +40,13 @@ public final class EntrySerializer
{
int start = out.position();
int totalSize = out.getInt() - start;
Invariants.checkState(totalSize == out.remaining() + TypeSizes.INT_SIZE);
Invariants.checkState(totalSize == record.remaining() + fixedEntrySize(keySupport, userVersion) + variableEntrySize(hosts.size()));
Invariants.checkState(totalSize == TypeSizes.INT_SIZE + out.remaining());
Invariants.checkState(totalSize == headerSize(keySupport, userVersion) + record.remaining() + TypeSizes.INT_SIZE);
keySupport.serialize(key, out, userVersion);
out.putShort((short)hosts.size());
int fixedCrcPosition = out.position();
out.position(fixedCrcPosition + TypeSizes.INT_SIZE);
for (int host : hosts)
out.putInt(host);
int headerCrcPosition = out.position();
out.position(headerCrcPosition + TypeSizes.INT_SIZE);
int recordSize = record.remaining();
int recordEnd = out.position() + recordSize;
@ -63,7 +56,7 @@ public final class EntrySerializer
// update and write crcs
CRC32 crc = Crc.crc32();
out.position(start);
out.limit(fixedCrcPosition);
out.limit(headerCrcPosition);
crc.update(out);
out.limit(recordEnd);
out.putInt((int) crc.getValue());
@ -87,9 +80,9 @@ public final class EntrySerializer
Invariants.checkState(totalSize == from.remaining());
CRC32 crc = Crc.crc32();
int fixedSize = EntrySerializer.fixedEntrySize(keySupport, userVersion);
int fixedCrc = readAndUpdateFixedCrc(crc, from, fixedSize);
validateCRC(crc, fixedCrc);
int headerSize = EntrySerializer.headerSize(keySupport, userVersion);
int headerCrc = readAndUpdateHeaderCrc(crc, from, headerSize);
validateCRC(crc, headerCrc);
int recordCrc = readAndUpdateRecordCrc(crc, from, start + totalSize);
validateCRC(crc, recordCrc);
@ -121,15 +114,15 @@ public final class EntrySerializer
return handleReadException(new EOFException(), from.limit(), syncedOffset);
{
int fixedSize = EntrySerializer.fixedEntrySize(keySupport, userVersion);
int fixedCrc = readAndUpdateFixedCrc(crc, from, fixedSize);
int headerSize = EntrySerializer.headerSize(keySupport, userVersion);
int headerCrc = readAndUpdateHeaderCrc(crc, from, headerSize);
try
{
validateCRC(crc, fixedCrc);
validateCRC(crc, headerCrc);
}
catch (IOException e)
{
return handleReadException(e, from.position() + fixedSize, syncedOffset);
return handleReadException(e, from.position() + headerSize, syncedOffset);
}
int recordCrc = readAndUpdateRecordCrc(crc, from, start + totalSize);
@ -151,26 +144,18 @@ public final class EntrySerializer
{
from.position(start + TypeSizes.INT_SIZE);
into.key = keySupport.deserialize(from, userVersion);
int hostCount = from.getShort();
from.position(from.position() + 4);
for (int i = 0; i < hostCount; i++)
{
int hostId = from.getInt();
into.hosts.add(hostId);
}
into.value = from;
into.userVersion = userVersion;
}
private static int readAndUpdateFixedCrc(CRC32 crc, ByteBuffer from, int fixedSize)
private static int readAndUpdateHeaderCrc(CRC32 crc, ByteBuffer from, int headerSize)
{
int fixedEnd = from.position() + fixedSize - TypeSizes.INT_SIZE;
int fixedCrc = from.getInt(fixedEnd);
from.limit(fixedEnd);
int headerEnd = from.position() + headerSize - TypeSizes.INT_SIZE;
int headerCrc = from.getInt(headerEnd);
from.limit(headerEnd);
crc.update(from);
return fixedCrc;
return headerCrc;
}
private static int readAndUpdateRecordCrc(CRC32 crc, ByteBuffer from, int limit)
@ -192,25 +177,17 @@ public final class EntrySerializer
return -1;
}
static <K> int fixedEntrySize(KeySupport<K> keySupport, int userVersion)
static <K> int headerSize(KeySupport<K> keySupport, int userVersion)
{
return keySupport.serializedSize(userVersion) // key/id
+ TypeSizes.SHORT_SIZE // host count
+ TypeSizes.INT_SIZE // total size
return TypeSizes.INT_SIZE // pointer to next entry
+ keySupport.serializedSize(userVersion) // key/id
+ TypeSizes.INT_SIZE; // CRC
}
static int variableEntrySize(int hostCount)
{
return TypeSizes.INT_SIZE * hostCount // hosts
+ TypeSizes.INT_SIZE; // CRC
}
public static final class EntryHolder<K>
{
public K key;
public ByteBuffer value;
public IntHashSet hosts = new IntHashSet();
public int userVersion;
@ -218,7 +195,6 @@ public final class EntrySerializer
{
key = null;
value = null;
hosts.clear();
}
}
}

31
src/java/org/apache/cassandra/journal/Journal.java
View File

@ -25,7 +25,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@ -74,9 +73,7 @@ import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue;
* A generic append-only journal with some special features:
* <p><ul>
* <li>Records can be looked up by key
* <li>Records can be tagged with multiple owner node ids
* <li>Records can be invalidated by their owner ids
* <li>Fully invalidated records get purged during segment compaction
* <li>Invalidated records get purged during segment compaction
* </ul><p>
*
* Type parameters:
@ -349,7 +346,7 @@ public class Journal<K, V> implements Shutdownable
public List<V> readAll(K id)
{
List<V> res = new ArrayList<>(2);
readAll(id, (segment, position, key, buffer, hosts, userVersion) -> {
readAll(id, (segment, position, key, buffer, userVersion) -> {
try (DataInputBuffer in = new DataInputBuffer(buffer, false))
{
res.add(valueSerializer.deserialize(key, in, userVersion));
@ -448,15 +445,14 @@ public class Journal<K, V> implements Shutdownable
*
* @param id user-provided record id, expected to roughly correlate with time and go up
* @param record the record to store
* @param hosts hosts expected to invalidate the record
*/
public void blockingWrite(K id, V record, Set<Integer> hosts)
public void blockingWrite(K id, V record)
{
try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get())
{
valueSerializer.serialize(id, record, dob, params.userVersion());
ActiveSegment<K, V>.Allocation alloc = allocate(dob.getLength(), hosts);
alloc.writeInternal(id, dob.unsafeGetBufferAndFlip(), hosts);
ActiveSegment<K, V>.Allocation alloc = allocate(dob.getLength());
alloc.writeInternal(id, dob.unsafeGetBufferAndFlip());
flusher.flushAndAwaitDurable(alloc);
}
catch (IOException e)
@ -474,20 +470,19 @@ public class Journal<K, V> implements Shutdownable
*
* @param id user-provided record id, expected to roughly correlate with time and go up
* @param record the record to store
* @param hosts hosts expected to invalidate the record
*/
public RecordPointer asyncWrite(K id, V record, Set<Integer> hosts)
public RecordPointer asyncWrite(K id, V record)
{
return asyncWrite(id, (out, userVersion) -> valueSerializer.serialize(id, record, out, userVersion), hosts);
return asyncWrite(id, (out, userVersion) -> valueSerializer.serialize(id, record, out, userVersion));
}
public RecordPointer asyncWrite(K id, Writer writer, Set<Integer> hosts)
public RecordPointer asyncWrite(K id, Writer writer)
{
try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get())
{
writer.write(dob, params.userVersion());
ActiveSegment<K, V>.Allocation alloc = allocate(dob.getLength(), hosts);
alloc.write(id, dob.unsafeGetBufferAndFlip(), hosts);
ActiveSegment<K, V>.Allocation alloc = allocate(dob.getLength());
alloc.write(id, dob.unsafeGetBufferAndFlip());
return flusher.flush(alloc);
}
catch (IOException e)
@ -497,12 +492,12 @@ public class Journal<K, V> implements Shutdownable
}
}
private ActiveSegment<K, V>.Allocation allocate(int entrySize, Set<Integer> hosts)
private ActiveSegment<K, V>.Allocation allocate(int entrySize)
{
ActiveSegment<K, V> segment = currentSegment;
ActiveSegment<K, V>.Allocation alloc;
while (null == (alloc = segment.allocate(entrySize, hosts)))
while (null == (alloc = segment.allocate(entrySize)))
{
if (entrySize >= (params.segmentSize() * 3) / 4)
throw new IllegalStateException("entrySize " + entrySize + " too large for a segmentSize of " + params.segmentSize());
@ -972,7 +967,7 @@ public class Journal<K, V> implements Shutdownable
break;
Invariants.checkState(next == readers.poll());
reader.accept(next.descriptor.timestamp, next.offset, next.key(), next.record(), next.hosts(), next.descriptor.userVersion);
reader.accept(next.descriptor.timestamp, next.offset, next.key(), next.record(), next.descriptor.userVersion);
if (next.advance())
readers.add(next);
else

95
src/java/org/apache/cassandra/journal/Metadata.java
View File

@ -19,15 +19,9 @@ package org.apache.cassandra.journal;
import java.io.EOFException;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.zip.CRC32;
import org.agrona.collections.Int2IntHashMap;
import org.agrona.collections.IntHashSet;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.utils.Crc;
@ -36,45 +30,32 @@ import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
/**
* Tracks and serializes the following information:
* - all the hosts with entries in the data segment and #of records each is tagged in;
* used for compaction prioritisation and to act in response to topology changes
* - total count of records in this segment file
* used for compaction prioritisation
*/
final class Metadata
{
private final Set<Integer> unmodifiableHosts;
private final Map<Integer, Integer> recordsPerHost;
private int fsyncLimit;
private volatile int recordsCount;
private static final AtomicIntegerFieldUpdater<Metadata> recordsCountUpdater =
AtomicIntegerFieldUpdater.newUpdater(Metadata.class, "recordsCount");
static Metadata create()
{
return new Metadata(new ConcurrentHashMap<>(), 0);
return new Metadata(0);
}
private Metadata(Map<Integer, Integer> recordsPerHost, int recordsCount)
private Metadata(int recordsCount)
{
this.recordsPerHost = recordsPerHost;
this.recordsCount = recordsCount;
this.unmodifiableHosts = Collections.unmodifiableSet(recordsPerHost.keySet());
}
void update(Set<Integer> hosts)
void update()
{
updateHosts(hosts);
incrementRecordsCount();
}
private void updateHosts(Set<Integer> hosts)
{
for (int host : hosts)
recordsPerHost.compute(host, (k, v) -> null == v ? 1 : v + 1);
}
int fsyncLimit()
{
return fsyncLimit;
@ -85,16 +66,6 @@ final class Metadata
recordsCountUpdater.incrementAndGet(this);
}
Set<Integer> hosts()
{
return unmodifiableHosts;
}
int count(int host)
{
return recordsPerHost.getOrDefault(host, 0);
}
int totalCount()
{
return recordsCount;
@ -103,64 +74,18 @@ final class Metadata
void write(DataOutputPlus out) throws IOException
{
CRC32 crc = Crc.crc32();
/* Write records count per host */
int size = recordsPerHost.size();
out.writeInt(size);
updateChecksumInt(crc, size);
out.writeInt((int) crc.getValue());
for (Map.Entry<Integer, Integer> entry : recordsPerHost.entrySet())
{
int host = entry.getKey();
int count = entry.getValue();
out.writeInt(host);
out.writeInt(count);
updateChecksumInt(crc, host);
updateChecksumInt(crc, count);
}
/* Write records count */
out.writeInt(recordsCount);
updateChecksumInt(crc, recordsCount);
out.writeInt((int) crc.getValue());
}
static Metadata read(DataInputPlus in) throws IOException
{
CRC32 crc = Crc.crc32();
/* Read records count per host */
int size = in.readInt();
updateChecksumInt(crc, size);
validateCRC(crc, in.readInt());
Int2IntHashMap recordsPerHost = new Int2IntHashMap(Integer.MIN_VALUE);
for (int i = 0; i < size; i++)
{
int host = in.readInt();
int count = in.readInt();
updateChecksumInt(crc, host);
updateChecksumInt(crc, count);
recordsPerHost.put(host, count);
}
/* Read records count */
int recordsCount = in.readInt();
updateChecksumInt(crc, recordsCount);
validateCRC(crc, in.readInt());
return new Metadata(recordsPerHost, recordsCount);
return new Metadata(recordsCount);
}
void persist(Descriptor descriptor)
@ -195,20 +120,12 @@ final class Metadata
static <K> Metadata rebuild(Descriptor descriptor, KeySupport<K> keySupport)
{
Int2IntHashMap recordsPerHost = new Int2IntHashMap(Integer.MIN_VALUE);
int recordsCount = 0;
try (StaticSegment.SequentialReader<K> reader = StaticSegment.sequentialReader(descriptor, keySupport, Integer.MAX_VALUE))
{
while (reader.advance())
{
// iterator is cached and reused by IntHashSet
IntHashSet.IntIterator hosts = reader.hosts().iterator();
while (hosts.hasNext())
recordsPerHost.merge(hosts.nextValue(), 1, Integer::sum);
++recordsCount;
}
}
catch (JournalReadError e)
{
@ -217,7 +134,7 @@ final class Metadata
throw e;
}
return new Metadata(recordsPerHost, recordsCount);
return new Metadata(recordsCount);
}
static <K> Metadata rebuildAndPersist(Descriptor descriptor, KeySupport<K> keySupport)

4
src/java/org/apache/cassandra/journal/RecordConsumer.java
View File

@ -19,10 +19,8 @@ package org.apache.cassandra.journal;
import java.nio.ByteBuffer;
import org.agrona.collections.IntHashSet;
@FunctionalInterface
public interface RecordConsumer<K>
{
void accept(long segment, int position, K key, ByteBuffer buffer, IntHashSet hosts, int userVersion);
void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion);
}

4
src/java/org/apache/cassandra/journal/Segment.java
View File

@ -92,7 +92,7 @@ public abstract class Segment<K, V> implements SelfRefCounted<Segment<K, V>>, Co
if (read(offset, size, into))
{
Invariants.checkState(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key);
consumer.accept(descriptor.timestamp, offset, id, into.value, into.hosts, descriptor.userVersion);
consumer.accept(descriptor.timestamp, offset, id, into.value, descriptor.userVersion);
return true;
}
return false;
@ -118,7 +118,7 @@ public abstract class Segment<K, V> implements SelfRefCounted<Segment<K, V>>, Co
Invariants.checkState(offset < prevOffset);
Invariants.checkState(read(offset, size, into), "Read should always return true");
Invariants.checkState(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key);
onEntry.accept(descriptor.timestamp, offset, into.key, into.value, into.hosts, into.userVersion);
onEntry.accept(descriptor.timestamp, offset, into.key, into.value, into.userVersion);
}
}

9
src/java/org/apache/cassandra/journal/StaticSegment.java
View File

@ -27,7 +27,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.agrona.collections.IntHashSet;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.Closeable;
@ -256,7 +255,7 @@ public final class StaticSegment<K, V> extends Segment<K, V>
{
while (reader.advance())
{
consumer.accept(descriptor.timestamp, reader.offset(), reader.key(), reader.record(), reader.hosts(), descriptor.userVersion);
consumer.accept(descriptor.timestamp, reader.offset(), reader.key(), reader.record(), descriptor.userVersion);
}
}
}
@ -322,12 +321,6 @@ public final class StaticSegment<K, V> extends Segment<K, V>
return holder.key;
}
public IntHashSet hosts()
{
ensureHasAdvanced();
return holder.hosts;
}
public ByteBuffer record()
{
ensureHasAdvanced();

11
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

@ -23,7 +23,6 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.NavigableMap;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
@ -100,8 +99,6 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
Invariants.checkState(MessagingService.current_version == MessagingService.VERSION_51, "Expected current version to be %d but given %d", MessagingService.VERSION_51, MessagingService.current_version);
}
private static final Set<Integer> SENTINEL_HOSTS = Collections.singleton(0);
static final ThreadLocal<byte[]> keyCRCBytes = ThreadLocal.withInitial(() -> new byte[JournalKeySupport.TOTAL_SIZE]);
private final Journal<JournalKey, Object> journal;
@ -295,7 +292,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
}
JournalKey key = new JournalKey(update.txnId, JournalKey.Type.COMMAND_DIFF, store);
RecordPointer pointer = journal.asyncWrite(key, diff, SENTINEL_HOSTS);
RecordPointer pointer = journal.asyncWrite(key, diff);
if (journalTable.shouldIndex(key)
&& diff.hasParticipants()
&& diff.after.route() != null)
@ -396,7 +393,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
private RecordPointer appendInternal(JournalKey key, Object write)
{
AccordJournalValueSerializers.FlyweightSerializer<Object, ?> serializer = (AccordJournalValueSerializers.FlyweightSerializer<Object, ?>) key.type.serializer;
return journal.asyncWrite(key, (out, userVersion) -> serializer.serialize(key, write, out, userVersion), SENTINEL_HOSTS);
return journal.asyncWrite(key, (out, userVersion) -> serializer.serialize(key, write, out, userVersion));
}
@VisibleForTesting
@ -454,12 +451,12 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
if (key.type != JournalKey.Type.COMMAND_DIFF)
{
// TODO (required): add "skip" for the key to avoid getting stuck
iter.readAllForKey(key, (segment, position, key1, buffer, hosts, userVersion) -> {});
iter.readAllForKey(key, (segment, position, key1, buffer, userVersion) -> {});
continue;
}
JournalKey finalKey = key;
iter.readAllForKey(key, (segment, position, local, buffer, hosts, userVersion) -> {
iter.readAllForKey(key, (segment, position, local, buffer, userVersion) -> {
Invariants.checkState(finalKey.equals(local));
try (DataInputBuffer in = new DataInputBuffer(buffer, false))
{

24
src/java/org/apache/cassandra/service/accord/AccordJournalTable.java
View File

@ -33,7 +33,6 @@ import org.slf4j.LoggerFactory;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.agrona.collections.IntHashSet;
import org.agrona.collections.LongHashSet;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.Operator;
@ -85,8 +84,6 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
{
private static final Logger logger = LoggerFactory.getLogger(AccordJournalTable.class);
private static final IntHashSet SENTINEL_HOSTS = new IntHashSet();
private final Journal<K, V> journal;
private final ColumnFamilyStore cfs;
@ -166,7 +163,7 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
}
@Override
public void accept(long segment, int position, K key, ByteBuffer buffer, IntHashSet hosts, int userVersion)
public void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion)
{
readBuffer(buffer, reader, userVersion);
}
@ -194,10 +191,10 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
}
@Override
public void accept(long segment, int position, K key, ByteBuffer buffer, IntHashSet hosts, int userVersion)
public void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion)
{
visit(segment);
super.accept(segment, position, key, buffer, hosts, userVersion);
super.accept(segment, position, key, buffer, userVersion);
}
}
@ -219,10 +216,10 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
}
@Override
public void accept(long segment, int position, K key, ByteBuffer buffer, IntHashSet hosts, int userVersion)
public void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion)
{
if (!tableRecordConsumer.visited(segment))
super.accept(segment, position, key, buffer, hosts, userVersion);
super.accept(segment, position, key, buffer, userVersion);
}
}
@ -383,10 +380,9 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
into.key = key;
into.value = row.getCell(recordColumn).buffer();
into.hosts = SENTINEL_HOSTS;
into.userVersion = Int32Type.instance.compose(row.getCell(versionColumn).buffer());
onEntry.accept(descriptor, position, into.key, into.value, into.hosts, into.userVersion);
onEntry.accept(descriptor, position, into.key, into.value, into.userVersion);
}
@SuppressWarnings("resource") // Auto-closeable iterator will release related resources
@ -434,9 +430,9 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
{
EntryHolder<K> into = new EntryHolder<>();
// TODO: use flyweight to avoid allocating extra lambdas?
readRow(key, partition.next(), into, (segment, position, key1, buffer, hosts, userVersion) -> {
readRow(key, partition.next(), into, (segment, position, key1, buffer, userVersion) -> {
visit(segment);
recordConsumer.accept(segment, position, key1, buffer, hosts, userVersion);
recordConsumer.accept(segment, position, key1, buffer, userVersion);
});
}
@ -501,9 +497,9 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
K tableKey = (K)tableIterator.key();
K journalKey = staticSegmentIterator.key();
if (journalKey != null && keySupport.compare(journalKey, key) == 0)
staticSegmentIterator.readAllForKey(key, (segment, position, key1, buffer, hosts, userVersion) -> {
staticSegmentIterator.readAllForKey(key, (segment, position, key1, buffer, userVersion) -> {
if (!tableIterator.visited(segment))
reader.accept(segment, position, key1, buffer, hosts, userVersion);
reader.accept(segment, position, key1, buffer, userVersion);
});
if (tableKey != null && keySupport.compare(tableKey, key) == 0)

3
test/simulator/test/org/apache/cassandra/simulator/test/AccordJournalSimulationTest.java
View File

@ -20,7 +20,6 @@ package org.apache.cassandra.simulator.test;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.zip.Checksum;
@ -96,7 +95,7 @@ public class AccordJournalSimulationTest extends SimulationTestBase
{
int finalI = i;
State.executor.submit(() -> {
RecordPointer ptr = State.journal.asyncWrite("test" + finalI, "test" + finalI, Collections.singleton(1));
RecordPointer ptr = State.journal.asyncWrite("test" + finalI, "test" + finalI);
State.journal.onDurable(ptr, State.latch::decrement);
});
}

12
test/unit/org/apache/cassandra/journal/JournalTest.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.journal;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Collections;
import org.junit.BeforeClass;
import org.junit.Test;
@ -50,8 +49,7 @@ public class JournalTest
directory.deleteRecursiveOnExit();
Journal<TimeUUID, Long> journal =
new Journal<>("TestJournal", directory, TestParams.INSTANCE, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop());
new Journal<>("TestJournal", directory, TestParams.INSTANCE, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop());
journal.start();
@ -60,10 +58,10 @@ public class JournalTest
TimeUUID id3 = nextTimeUUID();
TimeUUID id4 = nextTimeUUID();
journal.blockingWrite(id1, 1L, Collections.singleton(1));
journal.blockingWrite(id2, 2L, Collections.singleton(1));
journal.blockingWrite(id3, 3L, Collections.singleton(1));
journal.blockingWrite(id4, 4L, Collections.singleton(1));
journal.blockingWrite(id1, 1L);
journal.blockingWrite(id2, 2L);
journal.blockingWrite(id3, 3L);
journal.blockingWrite(id4, 4L);
assertEquals(1L, (long) journal.readLast(id1));
assertEquals(2L, (long) journal.readLast(id2));

54
test/unit/org/apache/cassandra/journal/MetadataTest.java
View File

@ -19,10 +19,6 @@ package org.apache.cassandra.journal;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import org.junit.Test;
@ -36,47 +32,25 @@ public class MetadataTest
@Test
public void testUpdate()
{
Random rng = new Random();
int host1 = rng.nextInt();
int host2 = host1 + 1;
int host3 = host2 + 1;
int host4 = host3 + 1;
int host5 = host4 + 1;
Metadata metadata = Metadata.create();
metadata.update(set(host1));
metadata.update(set(host2, host3));
metadata.update(set(host1, host4));
metadata.update(set(host1, host2, host3, host4));
metadata.update();
metadata.update();
metadata.update();
metadata.update();
assertEquals(set(host1, host2, host3, host4), metadata.hosts());
assertEquals(3, metadata.count(host1));
assertEquals(2, metadata.count(host2));
assertEquals(2, metadata.count(host3));
assertEquals(2, metadata.count(host4));
assertEquals(0, metadata.count(host5));
assertEquals(4, metadata.totalCount());
}
@Test
public void testWriteRead() throws IOException
{
Random rng = new Random();
int host1 = rng.nextInt();
int host2 = host1 + 1;
int host3 = host2 + 1;
int host4 = host3 + 1;
int host5 = host4 + 1;
Metadata metadata = Metadata.create();
metadata.update(set(host1));
metadata.update(set(host2, host3));
metadata.update(set(host1, host4));
metadata.update(set(host1, host2, host3, host4));
metadata.update();
metadata.update();
metadata.update();
metadata.update();
try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get())
{
@ -86,20 +60,8 @@ public class MetadataTest
try (DataInputBuffer in = new DataInputBuffer(serialized, false))
{
Metadata deserialized = Metadata.read(in);
assertEquals(set(host1, host2, host3, host4), deserialized.hosts());
assertEquals(3, deserialized.count(host1));
assertEquals(2, deserialized.count(host2));
assertEquals(2, deserialized.count(host3));
assertEquals(2, deserialized.count(host4));
assertEquals(0, deserialized.count(host5));
assertEquals(4, deserialized.totalCount());
}
}
}
private static Set<Integer> set(Integer... ids)
{
return new HashSet<>(Arrays.asList(ids));
}
}

78
test/unit/org/apache/cassandra/journal/SegmentTest.java
View File

@ -20,7 +20,6 @@ package org.apache.cassandra.journal;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.*;
import org.junit.Test;
@ -51,18 +50,6 @@ public class SegmentTest
ByteBuffer record3 = ByteBufferUtil.bytes("sample record 3");
ByteBuffer record4 = ByteBufferUtil.bytes("sample record 4");
Random rng = new Random();
int host1 = rng.nextInt();
int host2 = rng.nextInt();
int host3 = rng.nextInt();
int host4 = rng.nextInt();
Set<Integer> hosts1 = set(host1);
Set<Integer> hosts2 = set(host1, host2);
Set<Integer> hosts3 = set(host1, host2, host3);
Set<Integer> hosts4 = set(host4);
File directory = new File(Files.createTempDirectory(null));
directory.deleteRecursiveOnExit();
@ -70,32 +57,28 @@ public class SegmentTest
ActiveSegment<TimeUUID, ByteBuffer> segment = ActiveSegment.create(descriptor, params(), TimeUUIDKeySupport.INSTANCE);
segment.allocate(record1.remaining(), hosts1).write(id1, record1, hosts1);
segment.allocate(record2.remaining(), hosts2).write(id2, record2, hosts2);
segment.allocate(record3.remaining(), hosts3).write(id3, record3, hosts3);
segment.allocate(record4.remaining(), hosts4).write(id4, record4, hosts4);
segment.allocate(record1.remaining()).write(id1, record1);
segment.allocate(record2.remaining()).write(id2, record2);
segment.allocate(record3.remaining()).write(id3, record3);
segment.allocate(record4.remaining()).write(id4, record4);
// read all 4 entries by id and compare with originals
EntrySerializer.EntryHolder<TimeUUID> holder = new EntrySerializer.EntryHolder<>();
segment.readLast(id1, holder);
assertEquals(id1, holder.key);
assertEquals(hosts1, holder.hosts);
assertEquals(record1, holder.value);
segment.readLast(id2, holder);
assertEquals(id2, holder.key);
assertEquals(hosts2, holder.hosts);
assertEquals(record2, holder.value);
segment.readLast(id3, holder);
assertEquals(id3, holder.key);
assertEquals(hosts3, holder.hosts);
assertEquals(record3, holder.value);
segment.readLast(id4, holder);
assertEquals(id4, holder.key);
assertEquals(hosts4, holder.hosts);
assertEquals(record4, holder.value);
}
@ -114,18 +97,6 @@ public class SegmentTest
ByteBuffer record3 = ByteBufferUtil.bytes("sample record 3");
ByteBuffer record4 = ByteBufferUtil.bytes("sample record 4");
Random rng = new Random();
int host1 = rng.nextInt();
int host2 = rng.nextInt();
int host3 = rng.nextInt();
int host4 = rng.nextInt();
Set<Integer> hosts1 = set(host1);
Set<Integer> hosts2 = set(host1, host2);
Set<Integer> hosts3 = set(host1, host2, host3);
Set<Integer> hosts4 = set(host4);
File directory = new File(Files.createTempDirectory(null));
directory.deleteRecursiveOnExit();
@ -133,10 +104,10 @@ public class SegmentTest
ActiveSegment<TimeUUID, ByteBuffer> activeSegment = ActiveSegment.create(descriptor, params(), TimeUUIDKeySupport.INSTANCE);
activeSegment.allocate(record1.remaining(), hosts1).write(id1, record1, hosts1);
activeSegment.allocate(record2.remaining(), hosts2).write(id2, record2, hosts2);
activeSegment.allocate(record3.remaining(), hosts3).write(id3, record3, hosts3);
activeSegment.allocate(record4.remaining(), hosts4).write(id4, record4, hosts4);
activeSegment.allocate(record1.remaining()).write(id1, record1);
activeSegment.allocate(record2.remaining()).write(id2, record2);
activeSegment.allocate(record3.remaining()).write(id3, record3);
activeSegment.allocate(record4.remaining()).write(id4, record4);
activeSegment.close(null);
@ -147,22 +118,18 @@ public class SegmentTest
staticSegment.readLast(id1, holder);
assertEquals(id1, holder.key);
assertEquals(hosts1, holder.hosts);
assertEquals(record1, holder.value);
staticSegment.readLast(id2, holder);
assertEquals(id2, holder.key);
assertEquals(hosts2, holder.hosts);
assertEquals(record2, holder.value);
staticSegment.readLast(id3, holder);
assertEquals(id3, holder.key);
assertEquals(hosts3, holder.hosts);
assertEquals(record3, holder.value);
staticSegment.readLast(id4, holder);
assertEquals(id4, holder.key);
assertEquals(hosts4, holder.hosts);
assertEquals(record4, holder.value);
}
@ -179,18 +146,6 @@ public class SegmentTest
ByteBuffer record3 = ByteBufferUtil.bytes("sample record 3");
ByteBuffer record4 = ByteBufferUtil.bytes("sample record 4");
Random rng = new Random();
int host1 = rng.nextInt();
int host2 = rng.nextInt();
int host3 = rng.nextInt();
int host4 = rng.nextInt();
Set<Integer> hosts1 = set(host1);
Set<Integer> hosts2 = set(host1, host2);
Set<Integer> hosts3 = set(host1, host2, host3);
Set<Integer> hosts4 = set(host4);
File directory = new File(Files.createTempDirectory(null));
directory.deleteRecursiveOnExit();
@ -198,10 +153,10 @@ public class SegmentTest
ActiveSegment<TimeUUID, ByteBuffer> activeSegment = ActiveSegment.create(descriptor, params(), TimeUUIDKeySupport.INSTANCE);
activeSegment.allocate(record1.remaining(), hosts1).write(id1, record1, hosts1);
activeSegment.allocate(record2.remaining(), hosts2).write(id2, record2, hosts2);
activeSegment.allocate(record3.remaining(), hosts3).write(id3, record3, hosts3);
activeSegment.allocate(record4.remaining(), hosts4).write(id4, record4, hosts4);
activeSegment.allocate(record1.remaining()).write(id1, record1);
activeSegment.allocate(record2.remaining()).write(id2, record2);
activeSegment.allocate(record3.remaining()).write(id3, record3);
activeSegment.allocate(record4.remaining()).write(id4, record4);
Segment.Tidier tidier = (Segment.Tidier)activeSegment.selfRef().tidier();
tidier.executor = ImmediateExecutor.INSTANCE;
@ -215,32 +170,23 @@ public class SegmentTest
// read all 4 entries sequentially and compare with originals
assertTrue(reader.advance());
assertEquals(id1, reader.key());
assertEquals(hosts1, reader.hosts());
assertEquals(record1, reader.record());
assertTrue(reader.advance());
assertEquals(id2, reader.key());
assertEquals(hosts2, reader.hosts());
assertEquals(record2, reader.record());
assertTrue(reader.advance());
assertEquals(id3, reader.key());
assertEquals(hosts3, reader.hosts());
assertEquals(record3, reader.record());
assertTrue(reader.advance());
assertEquals(id4, reader.key());
assertEquals(hosts4, reader.hosts());
assertEquals(record4, reader.record());
assertFalse(reader.advance());
}
private static Set<Integer> set(Integer... ids)
{
return new HashSet<>(Arrays.asList(ids));
}
private static Params params()
{
return TestParams.INSTANCE;