svn commit: r1673557 - in /hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc: EncodedReaderImpl.java InStream.java

[se...@apache.org]

Author: sershe
Date: Tue Apr 14 21:30:24 2015
New Revision: 1673557

URL: http://svn.apache.org/r1673557
Log:
HIVE-10336 : LLAP: small improvement to unlocking cache data (Sergey Shelukhin)

Modified:
    hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java
    hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java

Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java?rev=1673557&r1=1673556&r2=1673557&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java Tue Apr 14 21:30:24 2015
@@ -387,7 +387,7 @@ public class EncodedReaderImpl implement
               long unlockUntilCOffset = sctx.offset + sctx.length;
               DiskRangeList lastCached = InStream.uncompressStream(fileId, stripeOffset, iter,
                   sctx.offset, sctx.offset + sctx.length, zcr, codec, bufferSize, cache,
-                  sctx.stripeLevelStream, unlockUntilCOffset);
+                  sctx.stripeLevelStream, unlockUntilCOffset, sctx.offset);
               if (lastCached != null) {
                 iter = lastCached;
               }
@@ -417,7 +417,8 @@ public class EncodedReaderImpl implement
             boolean isStartOfStream = sctx.bufferIter == null;
             DiskRangeList range = isStartOfStream ? iter : sctx.bufferIter;
             DiskRangeList lastCached = InStream.uncompressStream(fileId, stripeOffset, range,
-                cOffset, endCOffset, zcr, codec, bufferSize, cache, cb, unlockUntilCOffset);
+                cOffset, endCOffset, zcr, codec, bufferSize, cache, cb, unlockUntilCOffset,
+                sctx.offset);
             if (lastCached != null) {
               sctx.bufferIter = iter = lastCached; // Reset iter just to ensure it's valid
             }

Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java?rev=1673557&r1=1673556&r2=1673557&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java Tue Apr 14 21:30:24 2015
@@ -638,8 +638,8 @@ public abstract class InStream extends I
   // TODO: move to EncodedReaderImpl
   public static DiskRangeList uncompressStream(long fileId, long baseOffset, DiskRangeList start,
       long cOffset, long endCOffset, ZeroCopyReaderShim zcr, CompressionCodec codec,
-      int bufferSize, LowLevelCache cache, StreamBuffer streamBuffer, long unlockUntilCOffset)
-          throws IOException {
+      int bufferSize, LowLevelCache cache, StreamBuffer streamBuffer, long unlockUntilCOffset,
+      long streamStartOffset) throws IOException {
     if (streamBuffer.cacheBuffers == null) {
       streamBuffer.cacheBuffers = new ArrayList<LlapMemoryBuffer>();
     } else {
@@ -679,7 +679,7 @@ public abstract class InStream extends I
         if (DebugUtils.isTraceOrcEnabled()) {
           LOG.info("Adding an already-uncompressed buffer " + cc.buffer);
         }
-        ponderReleaseInitialRefcount(cache, unlockUntilCOffset, cc);
+        ponderReleaseInitialRefcount(cache, unlockUntilCOffset, streamStartOffset, cc);
         lastCached = cc;
         next = current.next;
       } else if (current instanceof IncompleteCb)  {
@@ -774,7 +774,7 @@ public abstract class InStream extends I
 
     // 7. It may happen that we only use new compression buffers once. Release initial refcounts.
     for (ProcCacheChunk chunk : toDecompress) {
-      ponderReleaseInitialRefcount(cache, unlockUntilCOffset, chunk);
+      ponderReleaseInitialRefcount(cache, unlockUntilCOffset, streamStartOffset, chunk);
     }
 
     return lastCached;
@@ -792,11 +792,28 @@ public abstract class InStream extends I
   }
 
   private static void ponderReleaseInitialRefcount(LowLevelCache cache,
-      long unlockUntilCOffset, TrackedCacheChunk cc) {
+      long unlockUntilCOffset, long streamStartOffset, TrackedCacheChunk cc) {
     if (cc.getEnd() > unlockUntilCOffset) return;
+    assert !cc.isReleased;
+    releaseInitialRefcount(cache, cc, false);
+    // Release all the previous buffers that we may not have been able to release due to reuse.
+    DiskRangeList prev = cc.prev;
+    while (true) {
+      if ((prev == null) || (prev.getEnd() <= streamStartOffset)
+          || !(prev instanceof TrackedCacheChunk)) break;
+      TrackedCacheChunk prevCc = (TrackedCacheChunk)prev;
+      if (prevCc.isReleased) break;
+      releaseInitialRefcount(cache, prevCc, true);
+      prev = prev.prev;
+    }
+  }
+
+  private static void releaseInitialRefcount(
+      LowLevelCache cache, TrackedCacheChunk cc, boolean isBacktracking) {
     // This is the last RG for which this buffer will be used. Remove the initial refcount
     if (DebugUtils.isTraceLockingEnabled()) {
-      LOG.info("Unlocking " + cc.buffer + " for the fetching thread");
+      LOG.info("Unlocking " + cc.buffer + " for the fetching thread"
+          + (isBacktracking ? "; backtracking" : ""));
     }
     cache.releaseBuffer(cc.buffer);
     cc.isReleased = true;