From 1aaf0bb336c7c8984b796d42b0d084cb661d779c Mon Sep 17 00:00:00 2001 From: Alex Petrov Date: Wed, 5 Feb 2025 20:34:44 +0100 Subject: [PATCH] Fix null field accounting Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20297 --- .gitmodules | 4 +- modules/accord | 2 +- .../org/apache/cassandra/journal/Journal.java | 8 ++- .../service/accord/AccordJournal.java | 65 ++++++++++--------- .../service/accord/AccordJournalTable.java | 24 +++++-- .../service/accord/AccordJournalBurnTest.java | 2 - .../accord/SimulatedAccordCommandStore.java | 6 +- 7 files changed, 63 insertions(+), 48 deletions(-) diff --git a/.gitmodules b/.gitmodules index 616dacf610..22d4a4fb41 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,4 +1,4 @@ [submodule "modules/accord"] path = modules/accord - url = https://github.com/apache/cassandra-accord.git - branch = trunk + url = https://github.com/ifesdjeen/cassandra-accord.git + branch = CASSANDRA-20297 diff --git a/modules/accord b/modules/accord index 9328b83f62..a2ac02b4d2 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 9328b83f623a7877249b5fb18fe49e2e78e8c3a9 +Subproject commit a2ac02b4d20bb6ab9078e70a0a43c17c00f4b0fc diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index 6f0fdabfd3..08626b441f 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -336,13 +336,15 @@ public class Journal implements Shutdownable return null; } - public void readAll(K id, RecordConsumer consumer, boolean asc) + public void readAll(K id, RecordConsumer consumer) { EntrySerializer.EntryHolder holder = new EntrySerializer.EntryHolder<>(); try (OpOrder.Group group = readOrder.start()) { - for (Segment segment : segments.get().allSorted(asc)) + for (Segment segment : segments.get().allSorted(false)) + { segment.readAll(id, holder, consumer); + } } } @@ -360,7 +362,7 @@ public class Journal implements Shutdownable // can only throw if serializer is buggy throw new RuntimeException(e); } - }, false); + }); return res; } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index fd435a90d7..9a266788e2 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -82,7 +82,7 @@ import static accord.impl.CommandChange.isChanged; import static accord.impl.CommandChange.isNull; import static accord.impl.CommandChange.nextSetField; import static accord.impl.CommandChange.setChanged; -import static accord.impl.CommandChange.setFieldIsNull; +import static accord.impl.CommandChange.setFieldIsNullAndChanged; import static accord.impl.CommandChange.toIterableSetFields; import static accord.impl.CommandChange.unsetIterable; import static accord.impl.CommandChange.validateFlags; @@ -254,35 +254,35 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier } @Override - public RedundantBefore loadRedundantBefore(int store) + public RedundantBefore loadRedundantBefore(int commandStoreId) { - IdentityAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, store), false); + IdentityAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, commandStoreId)); return accumulator.get(); } @Override - public NavigableMap loadBootstrapBeganAt(int store) + public NavigableMap loadBootstrapBeganAt(int commandStoreId) { - IdentityAccumulator> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, store), false); + IdentityAccumulator> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, commandStoreId)); return accumulator.get(); } @Override - public NavigableMap loadSafeToRead(int store) + public NavigableMap loadSafeToRead(int commandStoreId) { - IdentityAccumulator> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, store), false); + IdentityAccumulator> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, commandStoreId)); return accumulator.get(); } @Override - public CommandStores.RangesForEpoch loadRangesForEpoch(int store) + public CommandStores.RangesForEpoch loadRangesForEpoch(int commandStoreId) { - IdentityAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store), false); + IdentityAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, commandStoreId)); return accumulator.get(); } @Override - public void saveCommand(int store, CommandUpdate update, @Nullable Runnable onFlush) + public void saveCommand(int commandStoreId, CommandUpdate update, @Nullable Runnable onFlush) { Writer diff = Writer.make(update.before, update.after); if (diff == null) @@ -292,7 +292,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier return; } - JournalKey key = new JournalKey(update.txnId, JournalKey.Type.COMMAND_DIFF, store); + JournalKey key = new JournalKey(update.txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId); RecordPointer pointer = journal.asyncWrite(key, diff); if (journalTable.shouldIndex(key) && diff.hasParticipants() @@ -307,7 +307,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier @Override public Iterator replayTopologies() { - AccordJournalValueSerializers.MapAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.TOPOLOGY_UPDATE, 0), false); + AccordJournalValueSerializers.MapAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.TOPOLOGY_UPDATE, 0)); return accumulator.get().values().iterator(); } @@ -338,25 +338,25 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier @Override public DurableBefore load() { - DurableBeforeAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, 0), false); + DurableBeforeAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, 0)); return accumulator.get(); } }; } @Override - public void saveStoreState(int store, FieldUpdates fieldUpdates, Runnable onFlush) + public void saveStoreState(int commandStoreId, FieldUpdates fieldUpdates, Runnable onFlush) { RecordPointer pointer = null; // TODO: avoid allocating keys if (fieldUpdates.newRedundantBefore != null) - pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, store), fieldUpdates.newRedundantBefore); + pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, commandStoreId), fieldUpdates.newRedundantBefore); if (fieldUpdates.newBootstrapBeganAt != null) - pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, store), fieldUpdates.newBootstrapBeganAt); + pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, commandStoreId), fieldUpdates.newBootstrapBeganAt); if (fieldUpdates.newSafeToRead != null) - pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, store), fieldUpdates.newSafeToRead); + pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, commandStoreId), fieldUpdates.newSafeToRead); if (fieldUpdates.newRangesForEpoch != null) - pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store), fieldUpdates.newRangesForEpoch); + pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, commandStoreId), fieldUpdates.newRangesForEpoch); if (onFlush == null) return; @@ -371,7 +371,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier { JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId); Builder builder = new Builder(txnId, load); - journalTable.readAll(key, builder::deserializeNext, false); + journalTable.readAll(key, builder::deserializeNext); return builder; } @@ -381,13 +381,13 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier return loadDiffs(commandStoreId, txnId, Load.ALL); } - private BUILDER readAll(JournalKey key, boolean asc) + private BUILDER readAll(JournalKey key) { BUILDER builder = (BUILDER) key.type.serializer.mergerFor(key); // TODO: this can be further improved to avoid allocating lambdas AccordJournalValueSerializers.FlyweightSerializer serializer = (AccordJournalValueSerializers.FlyweightSerializer) key.type.serializer; // TODO (expected): for those where we store an image, read only the first entry we find in DESC order - journalTable.readAll(key, (in, userVersion) -> serializer.deserialize(key, builder, in, userVersion), asc); + journalTable.readAll(key, (in, userVersion) -> serializer.deserialize(key, builder, in, userVersion)); return builder; } @@ -544,6 +544,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier continue; } + switch (field) { case EXECUTE_AT: @@ -648,31 +649,33 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier public void deserializeNext(DataInputPlus in, int userVersion) throws IOException { Invariants.require(txnId != null); - int flags = in.readInt(); - Invariants.require(flags != 0); + int readFlags = in.readInt(); + Invariants.require(readFlags != 0); nextCalled = true; count++; - int iterable = toIterableSetFields(flags); + int iterable = toIterableSetFields(readFlags); while (iterable != 0) { Field field = nextSetField(iterable); + // Since we are iterating in reverse order, we skip the fields that were + // set by entries writter later (i.e. already read ones). if (isChanged(field, this.flags) || isNull(field, mask)) { - if (!isNull(field, flags)) - skip(field, in, userVersion); + if (!isNull(field, readFlags)) + skip(txnId, field, in, userVersion); iterable = unsetIterable(field, iterable); continue; } - this.flags = setChanged(field, this.flags); - if (isNull(field, flags)) + if (isNull(field, readFlags)) { - this.flags = setFieldIsNull(field, this.flags); + this.flags = setFieldIsNullAndChanged(field, this.flags); } else { + this.flags = setChanged(field, this.flags); deserialize(field, in, userVersion); } @@ -731,7 +734,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier } } - private void skip(Field field, DataInputPlus in, int userVersion) throws IOException + private static void skip(TxnId txnId, Field field, DataInputPlus in, int userVersion) throws IOException { switch (field) { @@ -777,7 +780,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier break; case RESULT: // TODO (expected): skip - result = ResultSerializers.result.deserialize(in, userVersion); + ResultSerializers.result.deserialize(in, userVersion); break; } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java index f4c96fa122..5b16752e32 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java @@ -162,10 +162,21 @@ public class AccordJournalTable implements RangeSearche this.reader = reader; } + private long prevSegment = -1; + private long prevPosition = -1; + @Override public void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion) { + Invariants.require(prevSegment == -1 || segment <= prevSegment, + "Records should always be iterated over in a reverse order, but %s was seen after %s", segment, prevSegment); + if (prevSegment != segment) + prevPosition = -1; + Invariants.require(prevPosition == -1 || position < prevPosition, + "Records should always be iterated over in a reverse order, but %s was seen after %s", position, prevPosition); readBuffer(buffer, reader, userVersion); + prevSegment = segment; + prevPosition = position; } } @@ -173,6 +184,7 @@ public class AccordJournalTable implements RangeSearche { protected LongHashSet visited = null; protected RecordConsumer delegate; + TableRecordConsumer(RecordConsumer delegate) { this.delegate = delegate; @@ -203,6 +215,7 @@ public class AccordJournalTable implements RangeSearche private final K key; private final TableRecordConsumer tableRecordConsumer; private final RecordConsumer delegate; + JournalAndTableRecordConsumer(K key, RecordConsumer reader) { this.key = key; @@ -218,7 +231,7 @@ public class AccordJournalTable implements RangeSearche @Override public void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion) { - if (!tableRecordConsumer.visited(segment)) + if (!tableRecordConsumer.visited(segment)) //TODO: don't need this anymore delegate.accept(segment, position, key, buffer, userVersion); } } @@ -339,15 +352,15 @@ public class AccordJournalTable implements RangeSearche *

* When reading from journal segments, skip descriptors that were read from the table. */ - public void readAll(K key, Reader reader, boolean asc) + public void readAll(K key, Reader reader) { - readAll(key, new RecordConsumerAdapter(reader), asc); + readAll(key, new RecordConsumerAdapter(reader)); } - public void readAll(K key, RecordConsumer reader, boolean asc) + public void readAll(K key, RecordConsumer reader) { JournalAndTableRecordConsumer consumer = new JournalAndTableRecordConsumer(key, reader); - journal.readAll(key, consumer, asc); + journal.readAll(key, consumer); consumer.readTable(); } @@ -382,7 +395,6 @@ public class AccordJournalTable implements RangeSearche long descriptor = LongType.instance.compose(ByteBuffer.wrap((byte[]) row.clustering().get(0))); int position = Int32Type.instance.compose(ByteBuffer.wrap((byte[]) row.clustering().get(1))); - into.key = key; into.value = row.getCell(recordColumn).buffer(); into.userVersion = Int32Type.instance.compose(row.getCell(versionColumn).buffer()); diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java index 7ed6a8cec0..75fdd8fea8 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java @@ -23,7 +23,6 @@ import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -99,7 +98,6 @@ public class AccordJournalBurnTest extends BurnTestBase } - @Ignore @Test public void testOne() { diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index abaed8349e..e0c1d126bf 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -452,15 +452,15 @@ public class SimulatedAccordCommandStore implements AutoCloseable } @Override - public void saveCommand(int store, CommandUpdate update, Runnable onFlush) + public void saveCommand(int commandStoreId, CommandUpdate update, Runnable onFlush) { - super.saveCommand(store, update, onFlush); + super.saveCommand(commandStoreId, update, onFlush); if (!update.after.txnId().domain().isRange()) return; Command after = update.after; Route route = after.participants().route(); if (route != null) - index.update(0, store, after.txnId(), route); + index.update(0, commandStoreId, after.txnId(), route); } @Override