HW 5 Шишигин Степан DWS ИТМО

src/main/java/ru/vk/itmo/test/shishiginstepan/dao/BinarySearchSSTable.java

@@ -1,8 +1,5 @@
 package ru.vk.itmo.test.shishiginstepan.dao;
 
-import ru.vk.itmo.dao.BaseEntry;
-import ru.vk.itmo.dao.Entry;
-
 import java.io.IOException;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
@@ -15,7 +12,7 @@
 import java.util.Iterator;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-public class BinarySearchSSTable implements SSTable<MemorySegment, Entry<MemorySegment>> {
+public class BinarySearchSSTable implements SSTable<MemorySegment, EntryWithTimestamp<MemorySegment>> {
     private long tableSize;
     private long indexSize;
     private final MemorySegment tableSegment;
@@ -74,7 +71,8 @@ private long searchEntryPosition(MemorySegment key, boolean exact) {
 
             long keyOffset = getKeyOffset(m);
             long valOffset = normalize(getValOffset(m));
-            long mismatch = MemorySegment.mismatch(key, 0, key.byteSize(), tableSegment, keyOffset, valOffset);
+            long keyEnd = valOffset - ValueLayout.JAVA_LONG_UNALIGNED.byteSize();
+            long mismatch = MemorySegment.mismatch(key, 0, key.byteSize(), tableSegment, keyOffset, keyEnd);
             if (mismatch == -1) {
                 return m;
             }
@@ -99,18 +97,20 @@ private long searchEntryPosition(MemorySegment key, boolean exact) {
     }
 
     @Override
-    public Entry<MemorySegment> get(MemorySegment key) {
+    public EntryWithTimestamp<MemorySegment> get(MemorySegment key) {
         MemorySegment val;
         long m = this.searchEntryPosition(key, true);
         if (m == -1) return null;
         long valOffset = getValOffset(m);
         long recordEnd = getRecordEnd(m);
+        long timestampOffset = getTimestampOffset(valOffset);
         val = valOffset < 0 ? null : tableSegment.asSlice(valOffset, recordEnd - valOffset);
-        return new BaseEntry<>(key, val);
+        long timestamp = tableSegment.get(ValueLayout.JAVA_LONG_UNALIGNED, timestampOffset);
+        return new EntryWithTimestamp<>(key, val, timestamp);
     }
 
     @Override
-    public Iterator<Entry<MemorySegment>> scan(MemorySegment keyFrom, MemorySegment keyTo) {
+    public Iterator<EntryWithTimestamp<MemorySegment>> scan(MemorySegment keyFrom, MemorySegment keyTo) {
         long startIndex;
         long endIndex;
         if (keyFrom == null) {
@@ -137,6 +137,10 @@ private long getKeyOffset(long recordIndex) {
         return indexSegment.get(ValueLayout.JAVA_LONG_UNALIGNED, recordIndex * Long.BYTES * 2);
     }
 
+    private long getTimestampOffset(long valOffset) { // таймстемп - лонг который лежит прямо перед значением record`а.
+        return normalize(valOffset) - ValueLayout.JAVA_LONG_UNALIGNED.byteSize();
+    }
+
     private long getValOffset(long recordIndex) {
         return indexSegment.get(ValueLayout.JAVA_LONG_UNALIGNED, recordIndex * Long.BYTES * 2 + Long.BYTES);
     }
@@ -150,7 +154,7 @@ private long getRecordEnd(long recordIndex) {
         }
     }
 
-    private Iterator<Entry<MemorySegment>> iterator(long startEntryIndex, long endEntryIndex) {
+    private Iterator<EntryWithTimestamp<MemorySegment>> iterator(long startEntryIndex, long endEntryIndex) {
         return new Iterator<>() {
             long currentEntryIndex = startEntryIndex;
 
@@ -160,12 +164,13 @@ public boolean hasNext() {
             }
 
             @Override
-            public Entry<MemorySegment> next() {
+            public EntryWithTimestamp<MemorySegment> next() {
                 var keyOffset = getKeyOffset(currentEntryIndex);
                 var valOffset = getValOffset(currentEntryIndex);
+                var timeStampOffset = getTimestampOffset(valOffset);
                 long nextOffset = getRecordEnd(currentEntryIndex);
                 this.currentEntryIndex++;
-                return new BaseEntry<>(
+                return new EntryWithTimestamp<>(
                         tableSegment.asSlice(keyOffset, normalize(valOffset) - keyOffset),
                         valOffset < 0
                                 ?
@@ -174,7 +179,8 @@ public Entry<MemorySegment> next() {
                                 tableSegment.asSlice(
                                         normalize(valOffset),
                                         nextOffset - normalize(valOffset)
-                                )
+                                ),
+                        tableSegment.get(ValueLayout.JAVA_LONG_UNALIGNED, timeStampOffset)
                 );
             }
         };

src/main/java/ru/vk/itmo/test/shishiginstepan/dao/BinarySearchSSTableWriter.java

@@ -1,7 +1,5 @@
 package ru.vk.itmo.test.shishiginstepan.dao;
 
-import ru.vk.itmo.dao.Entry;
-
 import java.io.IOException;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
@@ -29,7 +27,7 @@ private BinarySearchSSTableWriter() {
     }
 
     public static BinarySearchSSTable writeSSTable(
-            Iterable<Entry<MemorySegment>> entries,
+            Iterable<EntryWithTimestamp<MemorySegment>> entries,
             Path path,
             int id,
             Arena arena
@@ -52,7 +50,8 @@ public static BinarySearchSSTable writeSSTable(
             if (entry.value() != null) {
                 dataSize += entry.value().byteSize();
             }
-            indexSize += Long.BYTES * 2;
+            indexSize += Long.BYTES * 2; // размер лонгов которые оффсеты ключа, значения
+            dataSize += Long.BYTES; // размер лонга который будет таймстемпом в таблице
         }
         try (var fileChannel = FileChannel.open(
                 tempSSTPath.get(),
@@ -99,7 +98,7 @@ public static BinarySearchSSTable writeSSTable(
     }
 
     private static void writeEntries(
-            Iterable<Entry<MemorySegment>> entries,
+            Iterable<EntryWithTimestamp<MemorySegment>> entries,
             MemorySegment tableSegment,
             MemorySegment indexSegment
     ) {
@@ -111,6 +110,10 @@ private static void writeEntries(
 
             MemorySegment.copy(entry.key(), 0, tableSegment, tableOffset, entry.key().byteSize());
             tableOffset += entry.key().byteSize();
+
+            tableSegment.set(ValueLayout.JAVA_LONG_UNALIGNED, tableOffset, entry.timestamp());
+            tableOffset += ValueLayout.JAVA_LONG_UNALIGNED.byteSize();
+
             if (entry.value() == null) {
                 indexSegment.set(ValueLayout.JAVA_LONG_UNALIGNED, indexOffset, tombstone(tableOffset));
                 indexOffset += ValueLayout.JAVA_LONG_UNALIGNED.byteSize();

src/main/java/ru/vk/itmo/test/shishiginstepan/dao/EntryWithTimestamp.java

@@ -0,0 +1,10 @@
+package ru.vk.itmo.test.shishiginstepan.dao;
+
+import ru.vk.itmo.dao.Entry;
+
+public record EntryWithTimestamp<D>(D key, D value, Long timestamp) implements Entry<D> {
+    @Override
+    public String toString() {
+        return "{" + key + ":" + value + "}" + " timestamp: " + timestamp;
+    }
+}

src/main/java/ru/vk/itmo/test/shishiginstepan/dao/InMemDaoImpl.java

@@ -1,7 +1,6 @@
 package ru.vk.itmo.test.shishiginstepan.dao;
 
 import ru.vk.itmo.dao.Dao;
-import ru.vk.itmo.dao.Entry;
 
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.ValueLayout;
@@ -24,7 +23,7 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-public class InMemDaoImpl implements Dao<MemorySegment, Entry<MemorySegment>> {
+public class InMemDaoImpl implements Dao<MemorySegment, EntryWithTimestamp<MemorySegment>> {
     private final ExecutorService executor;
     private final Lock flushLock = new ReentrantLock();
 
@@ -54,13 +53,13 @@ public class InMemDaoImpl implements Dao<MemorySegment, Entry<MemorySegment>> {
 
     private final long memStorageLimit;
 
-    private final AtomicReference<ConcurrentNavigableMap<MemorySegment, Entry<MemorySegment>>> tempStorage =
-            new AtomicReference<>(
-                    new ConcurrentSkipListMap<>(
-                            keyComparator
-                    )
-            );
-    private final AtomicReference<ConcurrentNavigableMap<MemorySegment, Entry<MemorySegment>>> memStorage =
+    private final AtomicReference<ConcurrentNavigableMap<MemorySegment, EntryWithTimestamp<MemorySegment>>>
+            tempStorage = new AtomicReference<>(
+                new ConcurrentSkipListMap<>(
+                        keyComparator
+                )
+    );
+    private final AtomicReference<ConcurrentNavigableMap<MemorySegment, EntryWithTimestamp<MemorySegment>>> memStorage =
             new AtomicReference<>(
                     new ConcurrentSkipListMap<>(
                             keyComparator
@@ -98,9 +97,9 @@ public InMemDaoImpl() {
     }
 
     @Override
-    public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to) {
-        Iterator<Entry<MemorySegment>> memIterator;
-        Iterator<Entry<MemorySegment>> tempIterator;
+    public Iterator<EntryWithTimestamp<MemorySegment>> get(MemorySegment from, MemorySegment to) {
+        Iterator<EntryWithTimestamp<MemorySegment>> memIterator;
+        Iterator<EntryWithTimestamp<MemorySegment>> tempIterator;
         if (to == null && from == null) {
             memIterator = this.memStorage.get().values().iterator();
             tempIterator = this.tempStorage.get().values().iterator();
@@ -115,7 +114,7 @@ public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to)
             tempIterator = this.tempStorage.get().subMap(from, to).values().iterator();
         }
 
-        List<Iterator<Entry<MemorySegment>>> iterators = new ArrayList<>();
+        List<Iterator<EntryWithTimestamp<MemorySegment>>> iterators = new ArrayList<>();
         iterators.add(memIterator);
         iterators.add(tempIterator);
         persistentStorage.enrichWithPersistentIterators(from, to, iterators);
@@ -124,23 +123,24 @@ public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to)
         );
     }
 
+    // если этот метод возвращает могилу, это валидный объект Ентри с таймстемпом смерти (RIP 😔)
     @Override
-    public Entry<MemorySegment> get(MemorySegment key) {
-        Entry<MemorySegment> entry = this.memStorage.get().get(key);
+    public EntryWithTimestamp<MemorySegment> get(MemorySegment key) {
+        EntryWithTimestamp<MemorySegment> entry = this.memStorage.get().get(key);
         if (entry == null) {
             entry = this.tempStorage.get().get(key);
         }
         if (entry == null) {
             entry = persistentStorage.get(key);
         }
-        if (entry != null && entry.value() == null) {
-            return null;
+        if (entry == null) {
+            return new EntryWithTimestamp<>(key, null, 0L); // у пустых значений timestamp ноль
         }
         return entry;
     }
 
     @Override
-    public void upsert(Entry<MemorySegment> entry) {
+    public void upsert(EntryWithTimestamp<MemorySegment> entry) {
         this.memStorage.get().put(entry.key(), entry);
         this.memStorageSize.updateAndGet(
                 (size) -> size + (entry.key().byteSize() + (entry.value() == null ? 0 : entry.value().byteSize()))

src/main/java/ru/vk/itmo/test/shishiginstepan/dao/MergeIterator.java

@@ -9,7 +9,7 @@
 import java.util.List;
 import java.util.PriorityQueue;
 
-public class MergeIterator implements Iterator<Entry<MemorySegment>> {
+public class MergeIterator implements Iterator<EntryWithTimestamp<MemorySegment>> {
     private static final Comparator<MemorySegment> keyComparator = (o1, o2) -> {
         long mismatch = o1.mismatch(o2);
         if (mismatch == -1) {
@@ -33,12 +33,12 @@ public class MergeIterator implements Iterator<Entry<MemorySegment>> {
     });
 
     private static class PeekIteratorWrapper implements Iterator<Entry<MemorySegment>> {
-        private Entry<MemorySegment> prefetched;
-        private final Iterator<Entry<MemorySegment>> iterator;
+        private EntryWithTimestamp<MemorySegment> prefetched;
+        private final Iterator<EntryWithTimestamp<MemorySegment>> iterator;
 
         private final int priority;
 
-        public PeekIteratorWrapper(Iterator<Entry<MemorySegment>> iterator, int priority) {
+        public PeekIteratorWrapper(Iterator<EntryWithTimestamp<MemorySegment>> iterator, int priority) {
             this.iterator = iterator;
             this.priority = priority;
         }
@@ -49,11 +49,11 @@ public boolean hasNext() {
         }
 
         @Override
-        public Entry<MemorySegment> next() {
+        public EntryWithTimestamp<MemorySegment> next() {
             if (this.prefetched == null) {
                 return this.iterator.next();
             } else {
-                Entry<MemorySegment> toReturn = this.prefetched;
+                EntryWithTimestamp<MemorySegment> toReturn = this.prefetched;
                 this.prefetched = null;
                 return toReturn;
             }
@@ -75,10 +75,10 @@ public void skip() {
         }
     }
 
-    public MergeIterator(List<Iterator<Entry<MemorySegment>>> iterators) {
+    public MergeIterator(List<Iterator<EntryWithTimestamp<MemorySegment>>> iterators) {
         // приоритет мержа будет определен порядком итераторов
         for (int i = 0; i < iterators.size(); i++) {
-            Iterator<Entry<MemorySegment>> iterator = iterators.get(i);
+            Iterator<EntryWithTimestamp<MemorySegment>> iterator = iterators.get(i);
             if (iterator.hasNext()) {
                 this.iterators.add(new PeekIteratorWrapper(iterator, i));
             }
@@ -92,9 +92,9 @@ public boolean hasNext() {
     }
 
     @Override
-    public Entry<MemorySegment> next() {
+    public EntryWithTimestamp<MemorySegment> next() {
         PeekIteratorWrapper nextIterator = iterators.remove();
-        Entry<MemorySegment> nextEntry = nextIterator.next();
+        EntryWithTimestamp<MemorySegment> nextEntry = nextIterator.next();
         if (nextIterator.hasNext()) {
             this.iterators.add(nextIterator);
         }

src/main/java/ru/vk/itmo/test/shishiginstepan/dao/PersistentStorage.java

@@ -1,7 +1,5 @@
 package ru.vk.itmo.test.shishiginstepan.dao;
 
-import ru.vk.itmo.dao.Entry;
-
 import java.io.IOException;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
@@ -59,7 +57,7 @@ public void close() {
      * Гарантирует что при успешном завершении записи на диск, SSTable с переданными в метод данными
      * сразу будет доступен для чтения в PersistentStorage.
      **/
-    public void store(Iterable<Entry<MemorySegment>> data, int id) {
+    public void store(Iterable<EntryWithTimestamp<MemorySegment>> data, int id) {
         BinarySearchSSTable newSSTable = BinarySearchSSTableWriter.writeSSTable(
                 data,
                 basePath,
@@ -69,10 +67,10 @@ public void store(Iterable<Entry<MemorySegment>> data, int id) {
         this.sstables.add(newSSTable);
     }
 
-    public Entry<MemorySegment> get(MemorySegment key) {
+    public EntryWithTimestamp<MemorySegment> get(MemorySegment key) {
         for (BinarySearchSSTable sstable : this.sstables) {
             if (sstable.closed.get()) continue;
-            Entry<MemorySegment> ssTableResult = sstable.get(key);
+            EntryWithTimestamp<MemorySegment> ssTableResult = sstable.get(key);
             if (ssTableResult != null) {
                 return ssTableResult;
             }
@@ -83,25 +81,27 @@ public Entry<MemorySegment> get(MemorySegment key) {
     public void enrichWithPersistentIterators(
             MemorySegment from,
             MemorySegment to,
-            List<Iterator<Entry<MemorySegment>>> iteratorsToEnrich
+            List<Iterator<EntryWithTimestamp<MemorySegment>>> iteratorsToEnrich
     ) {
         iteratorsToEnrich.addAll(getPersistentIterators(from, to));
     }
 
-    private List<Iterator<Entry<MemorySegment>>> getPersistentIterators(
+    private List<Iterator<EntryWithTimestamp<MemorySegment>>> getPersistentIterators(
             MemorySegment from,
             MemorySegment to
     ) {
-        List<Iterator<Entry<MemorySegment>>> iterators = new ArrayList<>(sstables.size() + 1);
+        List<Iterator<EntryWithTimestamp<MemorySegment>>> iterators = new ArrayList<>(sstables.size() + 1);
         for (var sstable : sstables) {
             if (sstable.closed.get()) continue;
             iterators.add(sstable.scan(from, to));
         }
         return iterators;
     }
 
-    private List<Iterator<Entry<MemorySegment>>> getCompactableIterators(List<BinarySearchSSTable> tablesToCompact) {
-        List<Iterator<Entry<MemorySegment>>> iteratorsToCompact = new ArrayList<>();
+    private List<Iterator<EntryWithTimestamp<MemorySegment>>> getCompactableIterators(
+            List<BinarySearchSSTable> tablesToCompact
+    ) {
+        List<Iterator<EntryWithTimestamp<MemorySegment>>> iteratorsToCompact = new ArrayList<>();
         for (var sstable : tablesToCompact) {
             if (sstable.closed.get()) continue;
             iteratorsToCompact.add(sstable.scan(null, null));