Detectect overlaping ranges in level 0 files. Fix issue dain#94.

pcmind · Jan 3, 2018 · 4928585 · 4928585
1 parent 47e34e1
commit 4928585
Show file tree

Hide file tree

Showing 5 changed files with 66 additions and 28 deletions.
diff --git a/leveldb/src/main/java/org/iq80/leveldb/impl/DbImpl.java b/leveldb/src/main/java/org/iq80/leveldb/impl/DbImpl.java
@@ -531,6 +531,7 @@ else if (!isManual && compaction.isTrivialMove()) {
             doCompactionWork(compactionState);
             compaction.close(); //release resources
             cleanupCompaction(compactionState);
+            deleteObsoleteFiles();
         }
         if (compaction != null) {
             compaction.close();
@@ -1339,20 +1340,7 @@ private void installCompactionResults(CompactionState compact)
             pendingOutputs.remove(output.getNumber());
         }
 
-        try {
-            versions.logAndApply(compact.compaction.getEdit(), mutex);
-            deleteObsoleteFiles();
-        }
-        catch (IOException e) { //todo fix the issue causing this exception
-            // Compaction failed for some reason.  Simply discard the work and try again later.
-
-            // Discard any files we may have created during this failed compaction
-            for (FileMetaData output : compact.outputs) {
-                File file = new File(databaseDir, Filename.tableFileName(output.getNumber()));
-                file.delete();
-            }
-            compact.outputs.clear();
-        }
+        versions.logAndApply(compact.compaction.getEdit(), mutex);
     }
 
     @VisibleForTesting

diff --git a/leveldb/src/main/java/org/iq80/leveldb/impl/Level.java b/leveldb/src/main/java/org/iq80/leveldb/impl/Level.java
@@ -130,8 +130,22 @@ public LookupResult get(LookupKey key, ReadStats readStats)
         return null;
     }
 
-    public boolean someFileOverlapsRange(Slice smallestUserKey, Slice largestUserKey)
+    public boolean someFileOverlapsRange(boolean disjointSortedFiles, Slice smallestUserKey, Slice largestUserKey)
     {
+        UserComparator userComparator = internalKeyComparator.getUserComparator();
+        if (!disjointSortedFiles) {
+            // Need to check against all files
+            for (FileMetaData file : files) {
+                if (afterFile(userComparator, smallestUserKey, file) ||
+                        beforeFile(userComparator, largestUserKey, file)) {
+                    // No overlap
+                }
+                else {
+                    return true;  // Overlap
+                }
+            }
+            return false;
+        }
         int index = 0;
         if (smallestUserKey != null) {
             InternalKey smallestInternalKey = new InternalKey(smallestUserKey, MAX_SEQUENCE_NUMBER, VALUE);
@@ -143,8 +157,21 @@ public boolean someFileOverlapsRange(Slice smallestUserKey, Slice largestUserKey
             return false;
         }
 
-        UserComparator userComparator = internalKeyComparator.getUserComparator();
-        return (largestUserKey == null || userComparator.compare(largestUserKey, files.get(index).getSmallest().getUserKey()) >= 0);
+        return !beforeFile(userComparator, largestUserKey, files.get(index));
+    }
+
+    private boolean beforeFile(UserComparator userComparator, Slice userKey, FileMetaData file)
+    {
+        // null userKey occurs after all keys and is therefore never before *f
+        return (userKey != null &&
+                userComparator.compare(userKey, file.getSmallest().getUserKey()) < 0);
+    }
+
+    private boolean afterFile(UserComparator userComparator, Slice userKey, FileMetaData file)
+    {
+        // NULL user_key occurs before all keys and is therefore never after *f
+        return (userKey != null &&
+                userComparator.compare(userKey, file.getLargest().getUserKey()) > 0);
     }
 
     @VisibleForTesting

diff --git a/leveldb/src/main/java/org/iq80/leveldb/impl/Level0.java b/leveldb/src/main/java/org/iq80/leveldb/impl/Level0.java
@@ -108,8 +108,22 @@ public LookupResult get(LookupKey key, ReadStats readStats)
         return null;
     }
 
-    public boolean someFileOverlapsRange(Slice smallestUserKey, Slice largestUserKey)
+    public boolean someFileOverlapsRange(boolean disjointSortedFiles, Slice smallestUserKey, Slice largestUserKey)
     {
+        UserComparator userComparator = internalKeyComparator.getUserComparator();
+        if (!disjointSortedFiles) {
+            // Need to check against all files
+            for (FileMetaData file : files) {
+                if (afterFile(userComparator, smallestUserKey, file) ||
+                        beforeFile(userComparator, largestUserKey, file)) {
+                    // No overlap
+                }
+                else {
+                    return true;  // Overlap
+                }
+            }
+            return false;
+        }
         int index = 0;
         if (smallestUserKey != null) {
             InternalKey smallestInternalKey = new InternalKey(smallestUserKey, MAX_SEQUENCE_NUMBER, VALUE);
@@ -121,17 +135,26 @@ public boolean someFileOverlapsRange(Slice smallestUserKey, Slice largestUserKey
             return false;
         }
 
-        UserComparator userComparator = internalKeyComparator.getUserComparator();
-        return (largestUserKey == null || userComparator.compare(largestUserKey, files.get(index).getSmallest().getUserKey()) >= 0);
+        return !beforeFile(userComparator, largestUserKey, files.get(index));
+    }
+
+    private boolean beforeFile(UserComparator userComparator, Slice userKey, FileMetaData file)
+    {
+        // null userKey occurs after all keys and is therefore never before *f
+        return (userKey != null &&
+                userComparator.compare(userKey, file.getSmallest().getUserKey()) < 0);
+    }
+
+    private boolean afterFile(UserComparator userComparator, Slice userKey, FileMetaData file)
+    {
+        // NULL user_key occurs before all keys and is therefore never after *f
+        return (userKey != null &&
+                userComparator.compare(userKey, file.getLargest().getUserKey()) > 0);
     }
 
     @VisibleForTesting
     int findFile(InternalKey targetKey)
     {
-        if (files.isEmpty()) {
-            return files.size();
-        }
-
         // todo replace with Collections.binarySearch
         int left = 0;
         int right = files.size();

diff --git a/leveldb/src/main/java/org/iq80/leveldb/impl/Version.java b/leveldb/src/main/java/org/iq80/leveldb/impl/Version.java
@@ -176,7 +176,7 @@ int pickLevelForMemTableOutput(Slice smallestUserKey, Slice largestUserKey)
             // Push to next level if there is no overlap in next level,
             // and the #bytes overlapping in the level after that are limited.
             InternalKey start = new InternalKey(smallestUserKey, MAX_SEQUENCE_NUMBER, ValueType.VALUE);
-            InternalKey limit = new InternalKey(largestUserKey, 0, ValueType.VALUE);
+            InternalKey limit = new InternalKey(largestUserKey, 0, ValueType.DELETION);
             while (level < MAX_MEM_COMPACT_LEVEL) {
                 if (overlapInLevel(level + 1, smallestUserKey, largestUserKey)) {
                     break;
@@ -196,9 +196,9 @@ public boolean overlapInLevel(int level, Slice smallestUserKey, Slice largestUse
         checkPositionIndex(level, levels.size(), "Invalid level");
 
         if (level == 0) {
-            return level0.someFileOverlapsRange(smallestUserKey, largestUserKey);
+            return level0.someFileOverlapsRange(false, smallestUserKey, largestUserKey);
         }
-        return levels.get(level - 1).someFileOverlapsRange(smallestUserKey, largestUserKey);
+        return levels.get(level - 1).someFileOverlapsRange(true, smallestUserKey, largestUserKey);
     }
 
     public int numberOfLevels()

diff --git a/leveldb/src/test/java/org/iq80/leveldb/impl/VersionSetTest.java b/leveldb/src/test/java/org/iq80/leveldb/impl/VersionSetTest.java
@@ -65,7 +65,7 @@ boolean overlaps(String smallest, String largest)
     {
         Slice s = smallest != null ? TestUtils.asciiToSlice(smallest) : null;
         Slice l = largest != null ? TestUtils.asciiToSlice(largest) : null;
-        return newLevel().someFileOverlapsRange(s, l);
+        return newLevel().someFileOverlapsRange(true, s, l);
     }
 
     @Test