tez-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jeag...@apache.org
Subject tez git commit: TEZ-3813. Reduce Object size of MemoryFetchedInput for large jobs (Muhammad Samir Khan via jeagles)
Date Wed, 09 Aug 2017 22:40:16 GMT
Repository: tez
Updated Branches:
  refs/heads/master 8dcf8a121 -> 1061cf5c3


TEZ-3813. Reduce Object size of MemoryFetchedInput for large jobs (Muhammad Samir Khan via
jeagles)


Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/1061cf5c
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/1061cf5c
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/1061cf5c

Branch: refs/heads/master
Commit: 1061cf5c310f954ffbf96a3295325b5e2eeb5d9f
Parents: 8dcf8a1
Author: Jonathan Eagles <jeagles@yahoo-inc.com>
Authored: Wed Aug 9 17:40:09 2017 -0500
Committer: Jonathan Eagles <jeagles@yahoo-inc.com>
Committed: Wed Aug 9 17:40:09 2017 -0500

----------------------------------------------------------------------
 tez-runtime-library/findbugs-exclude.xml        |  6 +++
 .../common/readers/UnorderedKVReader.java       |  4 +-
 .../common/shuffle/DiskFetchedInput.java        | 44 +++++++++------
 .../library/common/shuffle/FetchedInput.java    | 56 +++++++++++--------
 .../runtime/library/common/shuffle/Fetcher.java |  2 +-
 .../common/shuffle/LocalDiskFetchedInput.java   | 43 +++++++++------
 .../common/shuffle/MemoryFetchedInput.java      | 57 +++++++++++++-------
 .../common/shuffle/impl/ShuffleManager.java     | 12 ++++-
 .../impl/SimpleFetchedInputAllocator.java       |  8 +--
 .../common/readers/TestUnorderedKVReader.java   |  2 +-
 .../library/common/shuffle/TestFetcher.java     |  3 +-
 .../common/shuffle/impl/TestShuffleManager.java | 12 ++++-
 12 files changed, 166 insertions(+), 83 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/findbugs-exclude.xml
----------------------------------------------------------------------
diff --git a/tez-runtime-library/findbugs-exclude.xml b/tez-runtime-library/findbugs-exclude.xml
index 24fe93e..9e55221 100644
--- a/tez-runtime-library/findbugs-exclude.xml
+++ b/tez-runtime-library/findbugs-exclude.xml
@@ -207,4 +207,10 @@
     <Field name="MAX_BUFFER_SIZE"/>
     <Bug pattern="MS_SHOULD_BE_FINAL"/>
   </Match>
+
+  <Match>
+    <Class name="org.apache.tez.runtime.library.common.shuffle.MemoryFetchedInput"/>
+    <Method name="getBytes"/>
+    <Bug pattern="EI_EXPOSE_REP"/>
+  </Match>
 </FindBugsFilter>

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/readers/UnorderedKVReader.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/readers/UnorderedKVReader.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/readers/UnorderedKVReader.java
index f4400db..60f70ac 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/readers/UnorderedKVReader.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/readers/UnorderedKVReader.java
@@ -221,10 +221,10 @@ public class UnorderedKVReader<K, V> extends KeyValueReader {
       MemoryFetchedInput mfi = (MemoryFetchedInput) fetchedInput;
 
       return new InMemoryReader(null, mfi.getInputAttemptIdentifier(),
-          mfi.getBytes(), 0, (int) mfi.getActualSize());
+          mfi.getBytes(), 0, (int) mfi.getSize());
     } else {
       return new IFile.Reader(fetchedInput.getInputStream(),
-          fetchedInput.getCompressedSize(), codec, null, null, ifileReadAhead,
+          fetchedInput.getSize(), codec, null, null, ifileReadAhead,
           ifileReadAheadLength, ifileBufferSize);
     }
   }

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/DiskFetchedInput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/DiskFetchedInput.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/DiskFetchedInput.java
index c873af7..5d1c037 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/DiskFetchedInput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/DiskFetchedInput.java
@@ -40,21 +40,33 @@ public class DiskFetchedInput extends FetchedInput {
   private final FileSystem localFS;
   private final Path tmpOutputPath;
   private final Path outputPath;
+  private final long size;
 
-  public DiskFetchedInput(long actualSize, long compressedSize,
+  public DiskFetchedInput(long compressedSize,
       InputAttemptIdentifier inputAttemptIdentifier,
       FetchedInputCallback callbackHandler, Configuration conf,
       LocalDirAllocator localDirAllocator, TezTaskOutputFiles filenameAllocator)
       throws IOException {
-    super(Type.DISK, actualSize, compressedSize, inputAttemptIdentifier, callbackHandler);
+    super(inputAttemptIdentifier, callbackHandler);
 
+    this.size = compressedSize;
     this.localFS = FileSystem.getLocal(conf).getRaw();
     this.outputPath = filenameAllocator.getInputFileForWrite(
-        this.inputAttemptIdentifier.getInputIdentifier(), this
-            .inputAttemptIdentifier.getSpillEventId(), actualSize);
+        this.getInputAttemptIdentifier().getInputIdentifier(), this
+            .getInputAttemptIdentifier().getSpillEventId(), this.size);
     // Files are not clobbered due to the id being appended to the outputPath in the tmpPath,
     // otherwise fetches for the same task but from different attempts would clobber each
other.
-    this.tmpOutputPath = outputPath.suffix(String.valueOf(id));
+    this.tmpOutputPath = outputPath.suffix(String.valueOf(getId()));
+  }
+
+  @Override
+  public Type getType() {
+    return Type.DISK;
+  }
+
+  @Override
+  public long getSize() {
+    return size;
   }
 
   @Override
@@ -68,7 +80,7 @@ public class DiskFetchedInput extends FetchedInput {
   }
 
   public final Path getInputPath() {
-    if (state == State.COMMITTED) {
+    if (isState(State.COMMITTED)) {
       return this.outputPath;
     }
     return this.tmpOutputPath;
@@ -76,8 +88,8 @@ public class DiskFetchedInput extends FetchedInput {
   
   @Override
   public void commit() throws IOException {
-    if (state == State.PENDING) {
-      state = State.COMMITTED;
+    if (isState(State.PENDING)) {
+      setState(State.COMMITTED);
       localFS.rename(tmpOutputPath, outputPath);
       notifyFetchComplete();
     }
@@ -85,8 +97,8 @@ public class DiskFetchedInput extends FetchedInput {
 
   @Override
   public void abort() throws IOException {
-    if (state == State.PENDING) {
-      state = State.ABORTED;
+    if (isState(State.PENDING)) {
+      setState(State.ABORTED);
       // TODO NEWTEZ Maybe defer this to container cleanup
       localFS.delete(tmpOutputPath, false);
       notifyFetchFailure();
@@ -96,10 +108,10 @@ public class DiskFetchedInput extends FetchedInput {
   @Override
   public void free() {
     Preconditions.checkState(
-        state == State.COMMITTED || state == State.ABORTED,
+        isState(State.COMMITTED) || isState(State.ABORTED),
         "FetchedInput can only be freed after it is committed or aborted");
-    if (state == State.COMMITTED) {
-      state = State.FREED;
+    if (isState(State.COMMITTED)) {
+      setState(State.FREED);
       try {
         // TODO NEWTEZ Maybe defer this to container cleanup
         localFS.delete(outputPath, false);
@@ -115,8 +127,8 @@ public class DiskFetchedInput extends FetchedInput {
   @Override
   public String toString() {
     return "DiskFetchedInput [outputPath=" + outputPath
-        + ", inputAttemptIdentifier=" + inputAttemptIdentifier
-        + ", actualSize=" + actualSize + ",compressedSize=" + compressedSize
-        + ", type=" + type + ", id=" + id + ", state=" + state + "]";
+        + ", inputAttemptIdentifier=" + getInputAttemptIdentifier()
+        + ", actualSize=" + getSize()
+        + ", type=" + getType() + ", id=" + getId() + ", state=" + getState() + "]";
   }
 }

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetchedInput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetchedInput.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetchedInput.java
index 3e740a0..8982c27 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetchedInput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetchedInput.java
@@ -42,38 +42,52 @@ public abstract class FetchedInput {
 
   private static AtomicInteger ID_GEN = new AtomicInteger(0);
 
-  protected InputAttemptIdentifier inputAttemptIdentifier;
-  protected final long actualSize;
-  protected final long compressedSize;
-  protected final Type type;
-  protected final FetchedInputCallback callback;
-  protected final int id;
-  protected State state;
-
-  public FetchedInput(Type type, long actualSize, long compressedSize,
-      InputAttemptIdentifier inputAttemptIdentifier,
+  private InputAttemptIdentifier inputAttemptIdentifier;
+  private final FetchedInputCallback callback;
+  private final int id;
+  private byte state;
+
+  protected FetchedInput(InputAttemptIdentifier inputAttemptIdentifier,
       FetchedInputCallback callbackHandler) {
-    this.type = type;
-    this.actualSize = actualSize;
-    this.compressedSize = compressedSize;
     this.inputAttemptIdentifier = inputAttemptIdentifier;
     this.callback = callbackHandler;
     this.id = ID_GEN.getAndIncrement();
-    this.state = State.PENDING;
+    this.state = (byte) State.PENDING.ordinal();
   }
 
-  public Type getType() {
-    return this.type;
+  public abstract Type getType();
+
+  protected boolean isState(State state) {
+    return this.state == (byte) state.ordinal();
   }
 
-  public long getActualSize() {
-    return this.actualSize;
+  protected void setState(State state) {
+    this.state = (byte) state.ordinal();
   }
-  
-  public long getCompressedSize() {
-    return this.compressedSize;
+
+  protected State getState() {
+    if (isState(State.PENDING)) {
+      return State.PENDING;
+    }
+    if (isState(State.COMMITTED)) {
+      return State.COMMITTED;
+    }
+    if (isState(State.ABORTED)) {
+      return State.ABORTED;
+    }
+    if (isState(State.FREED)) {
+      return State.FREED;
+    }
+    // Should not get here
+    return null;
   }
 
+  protected int getId() {
+    return this.id;
+  }
+
+  public abstract long getSize();
+
   public InputAttemptIdentifier getInputAttemptIdentifier() {
     return this.inputAttemptIdentifier;
   }

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
index bf8c83b..9f657e5 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
@@ -658,7 +658,7 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
           idxRecord = getTezIndexRecord(srcAttemptId, reduceId);
 
           fetchedInput = new LocalDiskFetchedInput(idxRecord.getStartOffset(),
-              idxRecord.getRawLength(), idxRecord.getPartLength(), srcAttemptId,
+              idxRecord.getPartLength(), srcAttemptId,
               getShuffleInputFileName(srcAttemptId.getPathComponent(), null),
               conf,
               new FetchedInputCallback() {

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/LocalDiskFetchedInput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/LocalDiskFetchedInput.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/LocalDiskFetchedInput.java
index 5c63961..0ae8f08 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/LocalDiskFetchedInput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/LocalDiskFetchedInput.java
@@ -39,18 +39,30 @@ public class LocalDiskFetchedInput extends FetchedInput {
   private final Path inputFile;
   private final FileSystem localFS;
   private final long startOffset;
+  private final long size;
 
-  public LocalDiskFetchedInput(long startOffset, long actualSize, long compressedSize,
+  public LocalDiskFetchedInput(long startOffset, long compressedSize,
                                InputAttemptIdentifier inputAttemptIdentifier, Path inputFile,
                                Configuration conf, FetchedInputCallback callbackHandler)
       throws IOException {
-    super(Type.DISK_DIRECT, actualSize, compressedSize, inputAttemptIdentifier, callbackHandler);
+    super(inputAttemptIdentifier, callbackHandler);
+    this.size = compressedSize;
     this.startOffset = startOffset;
     this.inputFile = inputFile;
     localFS = FileSystem.getLocal(conf);
   }
 
   @Override
+  public Type getType() {
+    return Type.DISK_DIRECT;
+  }
+
+  @Override
+  public long getSize() {
+    return size;
+  }
+
+  @Override
   public OutputStream getOutputStream() throws IOException {
       throw new IOException("Output Stream is not supported for " + this.toString());
   }
@@ -59,21 +71,21 @@ public class LocalDiskFetchedInput extends FetchedInput {
   public InputStream getInputStream() throws IOException {
     FSDataInputStream inputStream = localFS.open(inputFile);
     inputStream.seek(startOffset);
-    return new BoundedInputStream(inputStream, compressedSize);
+    return new BoundedInputStream(inputStream, getSize());
   }
 
   @Override
   public void commit() {
-    if (state == State.PENDING) {
-      state = State.COMMITTED;
+    if (isState(State.PENDING)) {
+      setState(State.COMMITTED);
       notifyFetchComplete();
     }
   }
 
   @Override
   public void abort() {
-    if (state == State.PENDING) {
-      state = State.ABORTED;
+    if (isState(State.PENDING)) {
+      setState(State.ABORTED);
       notifyFetchFailure();
     }
   }
@@ -81,10 +93,10 @@ public class LocalDiskFetchedInput extends FetchedInput {
   @Override
   public void free() {
     Preconditions.checkState(
-        state == State.COMMITTED || state == State.ABORTED,
+        isState(State.COMMITTED) || isState(State.ABORTED),
         "FetchedInput can only be freed after it is committed or aborted");
-    if (state == State.COMMITTED) { // ABORTED would have already called cleanup
-      state = State.FREED;
+    if (isState(State.COMMITTED)) { // ABORTED would have already called cleanup
+      setState(State.FREED);
       notifyFreedResource();
     }
   }
@@ -93,12 +105,11 @@ public class LocalDiskFetchedInput extends FetchedInput {
   public String toString() {
     return "LocalDiskFetchedInput [inputFile path =" + inputFile +
         ", offset" + startOffset +
-        ", actualSize=" + actualSize +
-        ", compressedSize=" + compressedSize +
-        ", inputAttemptIdentifier=" + inputAttemptIdentifier +
-        ", type=" + type +
-        ", id=" + id +
-        ", state=" + state + "]";
+        ", compressedSize=" + getSize() +
+        ", inputAttemptIdentifier=" + getInputAttemptIdentifier() +
+        ", type=" + getType() +
+        ", id=" + getId() +
+        ", state=" + getState() + "]";
   }
 
   @VisibleForTesting

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/MemoryFetchedInput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/MemoryFetchedInput.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/MemoryFetchedInput.java
index 78f1f3b..d0c3e77 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/MemoryFetchedInput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/MemoryFetchedInput.java
@@ -29,41 +29,54 @@ import com.google.common.base.Preconditions;
 
 public class MemoryFetchedInput extends FetchedInput {
 
-  private BoundedByteArrayOutputStream byteStream;
+  private byte[] byteArray;
 
-  public MemoryFetchedInput(long actualSize, long compressedSize,
+  public MemoryFetchedInput(long actualSize,
       InputAttemptIdentifier inputAttemptIdentifier,
       FetchedInputCallback callbackHandler) {
-    super(Type.MEMORY, actualSize, compressedSize, inputAttemptIdentifier, callbackHandler);
-    this.byteStream = new BoundedByteArrayOutputStream((int) actualSize);
+    super(inputAttemptIdentifier, callbackHandler);
+    this.byteArray = new byte[(int) actualSize];
+  }
+
+  @Override
+  public Type getType() {
+    return Type.MEMORY;
+  }
+
+  @Override
+  public long getSize() {
+    if (this.byteArray == null) {
+      return 0;
+    }
+    return this.byteArray.length;
   }
 
   @Override
   public OutputStream getOutputStream() {
-    return byteStream;
+    return new InMemoryBoundedByteArrayOutputStream(byteArray);
   }
 
   @Override
   public InputStream getInputStream() {
-    return new NonSyncByteArrayInputStream(byteStream.getBuffer());
+    return new NonSyncByteArrayInputStream(byteArray);
   }
 
   public byte[] getBytes() {
-    return byteStream.getBuffer();
+    return byteArray;
   }
   
   @Override
   public void commit() {
-    if (state == State.PENDING) {
-      state = State.COMMITTED;
+    if (isState(State.PENDING)) {
+      setState(State.COMMITTED);
       notifyFetchComplete();
     }
   }
 
   @Override
   public void abort() {
-    if (state == State.PENDING) {
-      state = State.ABORTED;
+    if (isState(State.PENDING)) {
+      setState(State.ABORTED);
       notifyFetchFailure();
     }
   }
@@ -71,20 +84,28 @@ public class MemoryFetchedInput extends FetchedInput {
   @Override
   public void free() {
     Preconditions.checkState(
-        state == State.COMMITTED || state == State.ABORTED,
+        isState(State.COMMITTED) || isState(State.ABORTED),
         "FetchedInput can only be freed after it is committed or aborted");
-    if (state == State.COMMITTED) { // ABORTED would have already called cleanup
-      state = State.FREED;
-      this.byteStream = null;
+    if (isState(State.COMMITTED)) { // ABORTED would have already called cleanup
+      setState(State.FREED);
       notifyFreedResource();
+      // Set this to null AFTER notifyFreedResource() so that getSize()
+      // returns the correct size
+      this.byteArray = null;
     }
   }
 
   @Override
   public String toString() {
     return "MemoryFetchedInput [inputAttemptIdentifier="
-        + inputAttemptIdentifier + ", actualSize=" + actualSize
-        + ", compressedSize=" + compressedSize + ", type=" + type + ", id="
-        + id + ", state=" + state + "]";
+        + getInputAttemptIdentifier() + ", size=" + getSize()
+        + ", type=" + getType() + ", id="
+        + getId() + ", state=" + getState() + "]";
+  }
+
+  private static class InMemoryBoundedByteArrayOutputStream extends BoundedByteArrayOutputStream
{
+    InMemoryBoundedByteArrayOutputStream(byte[] array) {
+      super(array, 0, array.length);
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
index 24fb12b..e142228 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
@@ -914,7 +914,17 @@ public class ShuffleManager implements FetcherCallback {
   static class NullFetchedInput extends FetchedInput {
 
     public NullFetchedInput(InputAttemptIdentifier inputAttemptIdentifier) {
-      super(Type.MEMORY, -1, -1, inputAttemptIdentifier, null);
+      super(inputAttemptIdentifier, null);
+    }
+
+    @Override
+    public Type getType() {
+      return Type.MEMORY;
+    }
+
+    @Override
+    public long getSize() {
+      return -1;
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
index f939cd1..6072c03 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
@@ -140,7 +140,7 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
       InputAttemptIdentifier inputAttemptIdentifier) throws IOException {
     if (actualSize > maxSingleShuffleLimit
         || this.usedMemory + actualSize > this.memoryLimit) {
-      return new DiskFetchedInput(actualSize, compressedSize,
+      return new DiskFetchedInput(compressedSize,
           inputAttemptIdentifier, this, conf, localDirAllocator,
           fileNameAllocator);
     } else {
@@ -149,7 +149,7 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
         LOG.info(srcNameTrimmed + ": " + "Used memory after allocating " + actualSize + "
: " +
             usedMemory);
       }
-      return new MemoryFetchedInput(actualSize, compressedSize, inputAttemptIdentifier, this);
+      return new MemoryFetchedInput(actualSize, inputAttemptIdentifier, this);
     }
   }
 
@@ -160,7 +160,7 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
 
     switch (type) {
     case DISK:
-      return new DiskFetchedInput(actualSize, compressedSize,
+      return new DiskFetchedInput(compressedSize,
           inputAttemptIdentifier, this, conf, localDirAllocator,
           fileNameAllocator);
     default:
@@ -197,7 +197,7 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
     case DISK:
       break;
     case MEMORY:
-      unreserve(fetchedInput.getActualSize());
+      unreserve(((MemoryFetchedInput) fetchedInput).getSize());
       break;
     default:
       throw new TezUncheckedException("InputType: " + fetchedInput.getType()

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/readers/TestUnorderedKVReader.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/readers/TestUnorderedKVReader.java
b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/readers/TestUnorderedKVReader.java
index c49a423..6fef944 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/readers/TestUnorderedKVReader.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/readers/TestUnorderedKVReader.java
@@ -96,7 +96,7 @@ public class TestUnorderedKVReader {
     createIFile(outputPath, 1);
 
     final LinkedList<LocalDiskFetchedInput> inputs = new LinkedList<LocalDiskFetchedInput>();
-    LocalDiskFetchedInput realFetchedInput = new LocalDiskFetchedInput(0, rawLen, compLen,
new
+    LocalDiskFetchedInput realFetchedInput = new LocalDiskFetchedInput(0, compLen, new
         InputAttemptIdentifier(0, 0), outputPath, defaultConf, new FetchedInputCallback()
{
       @Override
       public void fetchComplete(FetchedInput fetchedInput) {

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java
b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java
index b031154..db9c7af 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java
@@ -246,8 +246,7 @@ public class TestFetcher {
         SHUFFLE_INPUT_FILE_PREFIX + pathComponent);
     Assert.assertTrue("success callback fs", f.getLocalFS() instanceof LocalFileSystem);
     Assert.assertEquals("success callback filesystem", f.getStartOffset(), p * 10);
-    Assert.assertEquals("success callback raw size", f.getActualSize(), p * 1000);
-    Assert.assertEquals("success callback compressed size", f.getCompressedSize(), p * 100);
+    Assert.assertEquals("success callback compressed size", f.getSize(), p * 100);
     Assert.assertEquals("success callback input id", f.getInputAttemptIdentifier(), srcAttempId.expand(0));
     Assert.assertEquals("success callback type", f.getType(), FetchedInput.Type.DISK_DIRECT);
   }

http://git-wip-us.apache.org/repos/asf/tez/blob/1061cf5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java
b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java
index 23248ed..103f83d 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java
@@ -298,7 +298,17 @@ public class TestShuffleManager {
   static class TestFetchedInput extends FetchedInput {
 
     public TestFetchedInput(InputAttemptIdentifier inputAttemptIdentifier) {
-      super(Type.MEMORY, -1, -1, inputAttemptIdentifier, null);
+      super(inputAttemptIdentifier, null);
+    }
+
+    @Override
+    public long getSize() {
+      return -1;
+    }
+
+    @Override
+    public Type getType() {
+      return Type.MEMORY;
     }
 
     @Override


Mime
View raw message