Fix null field accounting

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20297

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/apache/cassandra-accord.git
+	url = https://github.com/ifesdjeen/cassandra-accord.git
-	branch = trunk
+	branch = CASSANDRA-20297

modules/accord

@@ -1 +1 @@
-Subproject commit 9328b83f623a7877249b5fb18fe49e2e78e8c3a9
+Subproject commit a2ac02b4d20bb6ab9078e70a0a43c17c00f4b0fc

src/java/org/apache/cassandra/journal/Journal.java

@@ -336,13 +336,15 @@ public class Journal<K, V> implements Shutdownable
         return null;
     }
 
-    public void readAll(K id, RecordConsumer<K> consumer, boolean asc)
+    public void readAll(K id, RecordConsumer<K> consumer)
     {
         EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
         try (OpOrder.Group group = readOrder.start())
         {
-            for (Segment<K, V> segment : segments.get().allSorted(asc))
+            for (Segment<K, V> segment : segments.get().allSorted(false))
+            {
                 segment.readAll(id, holder, consumer);
+            }
         }
     }
 
@@ -360,7 +362,7 @@ public class Journal<K, V> implements Shutdownable
                 // can only throw if serializer is buggy
                 throw new RuntimeException(e);
             }
-        }, false);
+        });
         return res;
     }
 

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<RedundantBefore> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, store), false);
+        IdentityAccumulator<RedundantBefore> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, commandStoreId));
         return accumulator.get();
     }
 
     @Override
-    public NavigableMap<TxnId, Ranges> loadBootstrapBeganAt(int store)
+    public NavigableMap<TxnId, Ranges> loadBootstrapBeganAt(int commandStoreId)
     {
-        IdentityAccumulator<NavigableMap<TxnId, Ranges>> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, store), false);
+        IdentityAccumulator<NavigableMap<TxnId, Ranges>> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, commandStoreId));
         return accumulator.get();
     }
 
     @Override
-    public NavigableMap<Timestamp, Ranges> loadSafeToRead(int store)
+    public NavigableMap<Timestamp, Ranges> loadSafeToRead(int commandStoreId)
     {
-        IdentityAccumulator<NavigableMap<Timestamp, Ranges>> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, store), false);
+        IdentityAccumulator<NavigableMap<Timestamp, Ranges>> 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<RangesForEpoch> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store), false);
+        IdentityAccumulator<RangesForEpoch> 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<TopologyUpdate> replayTopologies()
     {
-        AccordJournalValueSerializers.MapAccumulator<Long, TopologyUpdate> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.TOPOLOGY_UPDATE, 0), false);
+        AccordJournalValueSerializers.MapAccumulator<Long, TopologyUpdate> 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> BUILDER readAll(JournalKey key, boolean asc)
+    private <BUILDER> BUILDER readAll(JournalKey key)
     {
         BUILDER builder = (BUILDER) key.type.serializer.mergerFor(key);
         // TODO: this can be further improved to avoid allocating lambdas
         AccordJournalValueSerializers.FlyweightSerializer<?, BUILDER> serializer = (AccordJournalValueSerializers.FlyweightSerializer<?, BUILDER>) 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();
+            int readFlags = in.readInt();
-            Invariants.require(flags != 0);
+            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))
+                    if (!isNull(field, readFlags))
-                        skip(field, in, userVersion);
+                        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;
             }
         }

src/java/org/apache/cassandra/service/accord/AccordJournalTable.java

@@ -162,10 +162,21 @@ public class AccordJournalTable<K extends JournalKey, V> 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<K extends JournalKey, V> implements RangeSearche
     {
         protected LongHashSet visited = null;
         protected RecordConsumer<K> delegate;
+
         TableRecordConsumer(RecordConsumer<K> delegate)
         {
             this.delegate = delegate;
@@ -203,6 +215,7 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
         private final K key;
         private final TableRecordConsumer tableRecordConsumer;
         private final RecordConsumer<K> delegate;
+
         JournalAndTableRecordConsumer(K key, RecordConsumer<K> reader)
         {
             this.key = key;
@@ -218,7 +231,7 @@ public class AccordJournalTable<K extends JournalKey, V> 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<K extends JournalKey, V> implements RangeSearche
      * <p>
      * 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<K> reader, boolean asc)
+    public void readAll(K key, RecordConsumer<K> 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<K extends JournalKey, V> 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());

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()
     {

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