Качмар Евгений, ИТМО ФИТиП М33331, Reverse Iterator

src/main/java/ru/vk/itmo/test/kachmareugene/InMemoryDao.java

@@ -29,6 +29,10 @@ public InMemoryDao(Config conf) {
 
     @Override
     public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to) {
+        if (new MemSegComparatorNull().compare(from, to) > 0) {
+            return reverseIter(from, to);
+        }
+
         SortedMap<MemorySegment, Entry<MemorySegment>> dataSlice;
 
         if (from == null && to == null) {
@@ -44,6 +48,11 @@ public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to)
         return new SSTableIterator(dataSlice.values().iterator(), controller, from, to);
     }
 
+    private Iterator<Entry<MemorySegment>> reverseIter(MemorySegment fromRightSide, MemorySegment toLeftSide) {
+        return new SSTableIterator(getMemTable().reversed().values().iterator(),
+                controller, fromRightSide, toLeftSide, true, new MemSegComparatorNull().reversed());
+    }
+
     @Override
     public Entry<MemorySegment> get(MemorySegment key) {
         if (key == null) {

src/main/java/ru/vk/itmo/test/kachmareugene/IteratorUtils.java

@@ -0,0 +1,38 @@
+package ru.vk.itmo.test.kachmareugene;
+
+import ru.vk.itmo.Entry;
+
+import java.lang.foreign.MemorySegment;
+import java.util.SortedMap;
+
+public class IteratorUtils {
+    private IteratorUtils() {
+    }
+
+    public static void insertNew(SortedMap<MemorySegment, SSTableRowInfo> mp,
+                                 SSTablesController controller, SSTableRowInfo info, MemorySegment to) {
+        Entry<MemorySegment> kv = controller.getRow(info);
+
+        if (kv == null) {
+            return;
+        }
+
+        if (!mp.containsKey(kv.key())) {
+            mp.put(kv.key(), info);
+            return;
+        }
+        SSTableRowInfo old = mp.get(kv.key());
+
+        SSTableRowInfo oldInfo = old.ssTableInd > info.ssTableInd ? info : old;
+        SSTableRowInfo newInfo = old.ssTableInd < info.ssTableInd ? info : old;
+
+        mp.put(controller.getRow(newInfo).key(), newInfo);
+
+        // tail recursion
+        if (oldInfo.isReversedToIter) {
+            insertNew(mp, controller, controller.getPrevInfo(oldInfo, to), to);
+        } else {
+            insertNew(mp, controller, controller.getNextInfo(oldInfo, to), to);
+        }
+    }
+}

src/main/java/ru/vk/itmo/test/kachmareugene/SSTableIterator.java

@@ -14,51 +14,46 @@ public class SSTableIterator implements Iterator<Entry<MemorySegment>> {
 
     private final Iterator<Entry<MemorySegment>> memTableIterator;
     private final SSTablesController controller;
-    private final Comparator<MemorySegment> comp = new MemSegComparatorNull();
-    private final SortedMap<MemorySegment, SSTableRowInfo> mp = new TreeMap<>(comp);
+    private final Comparator<MemorySegment> comp;
+    private final SortedMap<MemorySegment, SSTableRowInfo> mp;
     private final MemorySegment from;
     private final MemorySegment to;
     private Entry<MemorySegment> head;
     private Entry<MemorySegment> keeper;
+    private boolean isReversed;
 
     public SSTableIterator(Iterator<Entry<MemorySegment>> it, SSTablesController controller,
                            MemorySegment from, MemorySegment to) {
-        memTableIterator = it;
+        this.memTableIterator = it;
         this.controller = controller;
+        this.comp = new MemSegComparatorNull();
+        this.mp = new TreeMap<>(comp);
 
         this.from = from;
         this.to = to;
 
         positioningIterator();
     }
 
-    private void insertNew(SSTableRowInfo info) {
-        Entry<MemorySegment> kv = controller.getRow(info);
-
-        if (kv == null) {
-            return;
-        }
-
-        if (!mp.containsKey(kv.key())) {
-            mp.put(kv.key(), info);
-            return;
-        }
-        SSTableRowInfo old = mp.get(kv.key());
-
-        SSTableRowInfo oldInfo = old.ssTableInd > info.ssTableInd ? info : old;
-        SSTableRowInfo newInfo = old.ssTableInd < info.ssTableInd ? info : old;
+    public SSTableIterator(Iterator<Entry<MemorySegment>> it, SSTablesController controller,
+                           MemorySegment from, MemorySegment to, boolean isReversed, Comparator<MemorySegment> comp) {
+        this.memTableIterator = it;
+        this.controller = controller;
+        this.comp = comp;
+        this.mp = new TreeMap<>(comp);
 
-        mp.put(controller.getRow(newInfo).key(), newInfo);
+        this.from = from;
+        this.to = to;
+        this.isReversed = isReversed;
 
-        // tail recursion
-        insertNew(controller.getNextInfo(oldInfo, to));
+        positioningIterator();
     }
 
     private void positioningIterator() {
-        List<SSTableRowInfo> rawData = controller.firstGreaterKeys(from);
+        List<SSTableRowInfo> rawData = controller.firstKeys(from, isReversed);
 
         for (var info : rawData) {
-            insertNew(info);
+            IteratorUtils.insertNew(mp, controller, info, to);
         }
     }
 
@@ -138,7 +133,13 @@ private void changeState() {
     }
 
     private void updateMp(MemorySegment key) {
-        insertNew(controller.getNextInfo(mp.remove(key), to));
+        if (isReversed) {
+            IteratorUtils.insertNew(mp, controller,
+                    controller.getPrevInfo(mp.remove(key), to), to);
+            return;
+        }
+        IteratorUtils.insertNew(mp, controller,
+                controller.getNextInfo(mp.remove(key), to), to);
     }
 
     private boolean isBetween(MemorySegment who) {
@@ -150,8 +151,7 @@ private boolean isBetween(MemorySegment who) {
     private Map.Entry<MemorySegment, SSTableRowInfo> getFirstMin() {
         Map.Entry<MemorySegment, SSTableRowInfo> minSStablesEntry = mp.firstEntry();
         while (!mp.isEmpty() && !isBetween(minSStablesEntry.getKey())) {
-            mp.remove(minSStablesEntry.getKey());
-            insertNew(controller.getNextInfo(minSStablesEntry.getValue(), to));
+            updateMp(minSStablesEntry.getKey());
             minSStablesEntry = mp.firstEntry();
         }
         return minSStablesEntry;

src/main/java/ru/vk/itmo/test/kachmareugene/SSTableRowInfo.java

@@ -7,6 +7,7 @@ public class SSTableRowInfo {
     long rowShift;
     private final long valueSize;
     int ssTableInd;
+    boolean isReversedToIter;
 
     public SSTableRowInfo(long keyOffset, long keySize, long valueOffset,
                           long valueSize, int ssTableInd, long rowShift) {
@@ -18,6 +19,17 @@ public SSTableRowInfo(long keyOffset, long keySize, long valueOffset,
         this.rowShift = rowShift;
     }
 
+    public SSTableRowInfo(long keyOffset, long keySize, long valueOffset,
+                          long valueSize, int ssTableInd, long rowShift, boolean isReversedToIter) {
+        this.keyOffset = keyOffset;
+        this.valueOffset = valueOffset;
+        this.keySize = keySize;
+        this.valueSize = valueSize;
+        this.ssTableInd = ssTableInd;
+        this.rowShift = rowShift;
+        this.isReversedToIter = isReversedToIter;
+    }
+
     public boolean isDeletedData() {
         return valueSize < 0;
     }

src/main/java/ru/vk/itmo/test/kachmareugene/SSTablesController.java

@@ -71,47 +71,34 @@ private List<Path> openFiles(Path dir, String fileNamePref, List<MemorySegment>
         }
     }
 
-    private boolean greaterThen(long keyOffset, long keySize,
-                                MemorySegment mapped, MemorySegment key) {
-
-        return segComp.compare(key, mapped.asSlice(keyOffset, keySize)) > 0;
-    }
-
-    //Gives offset for line in index file
-    private long searchKeyInFile(int ind, MemorySegment mapped, MemorySegment key) {
-        long l = -1;
-        long r = getNumberOfEntries(mapped);
-
-        while (r - l > 1) {
-            long mid = (l + r) / 2;
-            SSTableRowInfo info = createRowInfo(ind, mid);
-            if (greaterThen(info.keyOffset, info.keySize, mapped, key)) {
-                l = mid;
-            } else {
-                r = mid;
-            }
-        }
-        return r == getNumberOfEntries(mapped) ? -1 : r;
-    }
-
     //return - List ordered form the latest created sstable to the first.
-    public List<SSTableRowInfo> firstGreaterKeys(MemorySegment key) {
+    public List<SSTableRowInfo> firstKeys(MemorySegment key, boolean isReversed) {
         List<SSTableRowInfo> ans = new ArrayList<>();
 
         for (int i = ssTables.size() - 1; i >= 0; i--) {
             long entryIndexesLine = 0;
             if (key != null) {
-                entryIndexesLine = searchKeyInFile(i, ssTables.get(i), key);
+                if (isReversed) {
+                    entryIndexesLine = Utils.searchKeyInFileReversed(this, i,
+                            getNumberOfEntries(ssTables.get(i)),
+                            ssTables.get(i), key, segComp);
+                } else {
+                    entryIndexesLine = Utils.searchKeyInFile(this, i,
+                            getNumberOfEntries(ssTables.get(i)), ssTables.get(i), key, segComp);
+                }
              }
             if (entryIndexesLine < 0) {
                 continue;
             }
-            ans.add(createRowInfo(i, entryIndexesLine));
+            SSTableRowInfo row = createRowInfo(i, entryIndexesLine);
+            row.isReversedToIter = isReversed;
+
+            ans.add(row);
         }
         return ans;
     }
 
-    private SSTableRowInfo createRowInfo(int ind, final long rowIndex) {
+    public SSTableRowInfo createRowInfo(int ind, final long rowIndex) {
         long start = ssTables.get(ind).get(ValueLayout.JAVA_LONG_UNALIGNED, rowIndex * ONE_LINE_SIZE + Long.BYTES);
         long size = ssTables.get(ind).get(ValueLayout.JAVA_LONG_UNALIGNED, rowIndex * ONE_LINE_SIZE + Long.BYTES * 2);
 
@@ -122,7 +109,8 @@ private SSTableRowInfo createRowInfo(int ind, final long rowIndex) {
 
     public SSTableRowInfo searchInSStables(MemorySegment key) {
         for (int i = ssTables.size() - 1; i >= 0; i--) {
-            long ind = searchKeyInFile(i, ssTables.get(i), key);
+            long ind = Utils.searchKeyInFile(this, i,
+                    getNumberOfEntries(ssTables.get(i)), ssTables.get(i), key, segComp);
             if (ind >= 0) {
                 return createRowInfo(i, ind);
             }
@@ -160,6 +148,18 @@ public SSTableRowInfo getNextInfo(SSTableRowInfo info, MemorySegment maxKey) {
         return null;
     }
 
+    public SSTableRowInfo getPrevInfo(SSTableRowInfo info, MemorySegment minKey) {
+        for (long t = info.rowShift - 1; t >= 0; t--) {
+            var inf = createRowInfo(info.ssTableInd, t);
+
+            Entry<MemorySegment> row = getRow(inf);
+            if (segComp.compare(row.key(), minKey) > 0) {
+                return inf;
+            }
+        }
+        return null;
+    }
+
     private long getNumberOfEntries(MemorySegment memSeg) {
         return memSeg.get(ValueLayout.JAVA_LONG_UNALIGNED, 0);
     }

src/main/java/ru/vk/itmo/test/kachmareugene/Utils.java

@@ -72,4 +72,56 @@ public static MemorySegment getValueOrNull(Entry<MemorySegment> kv) {
         }
         return value;
     }
+
+    private static boolean greaterThen(Comparator<MemorySegment> segComp, long keyOffset, long keySize,
+                                       MemorySegment mapped, MemorySegment key) {
+
+        return segComp.compare(key, mapped.asSlice(keyOffset, keySize)) > 0;
+    }
+
+    private static boolean greaterEqThen(Comparator<MemorySegment> segComp, long keyOffset, long keySize,
+                                         MemorySegment mapped, MemorySegment key) {
+
+        return segComp.compare(key, mapped.asSlice(keyOffset, keySize)) >= 0;
+    }
+
+    //Gives offset for line in index file
+    public static long searchKeyInFile(SSTablesController controller,
+                                       int ind, long numberOfEntries,
+                                       MemorySegment mapped, MemorySegment key,
+                                       Comparator<MemorySegment> segComp) {
+        long l = -1;
+        long r = numberOfEntries;
+
+        while (r - l > 1) {
+            long mid = (l + r) / 2;
+            SSTableRowInfo info = controller.createRowInfo(ind, mid);
+            if (greaterThen(segComp, info.keyOffset, info.keySize, mapped, key)) {
+                l = mid;
+            } else {
+                r = mid;
+            }
+        }
+        return r == numberOfEntries ? -1 : r;
+    }
+
+    public static long searchKeyInFileReversed(SSTablesController controller,
+                                               int ind, long numberOfEntries,
+                                               MemorySegment mapped, MemorySegment key,
+                                               Comparator<MemorySegment> segComp) {
+        long l = -1;
+        long r = numberOfEntries;
+
+        while (r - l > 1) {
+            long mid = (l + r) / 2;
+            SSTableRowInfo info = controller.createRowInfo(ind, mid);
+            if (greaterEqThen(segComp, info.keyOffset, info.keySize, mapped, key)) {
+                l = mid;
+            } else {
+                r = mid;
+            }
+        }
+        return l;
+    }
+
 }