You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by go...@apache.org on 2014/04/25 23:04:05 UTC
git commit: TEZ-1073. RLE fast-forward merge for IFile (gopalv)
Repository: incubator-tez
Updated Branches:
refs/heads/master b51af35e1 -> 2efe086d8
TEZ-1073. RLE fast-forward merge for IFile (gopalv)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/2efe086d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/2efe086d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/2efe086d
Branch: refs/heads/master
Commit: 2efe086d86209755f5a78472e39a5711fdbd7017
Parents: b51af35
Author: Gopal V <go...@apache.org>
Authored: Fri Apr 25 13:58:20 2014 -0700
Committer: Gopal V <go...@apache.org>
Committed: Fri Apr 25 13:58:20 2014 -0700
----------------------------------------------------------------------
.../library/common/shuffle/impl/MergeManager.java | 7 ++++---
.../tez/runtime/library/common/sort/impl/IFile.java | 2 +-
.../runtime/library/common/sort/impl/TezMerger.java | 13 ++++++++++---
3 files changed, 15 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/2efe086d/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/MergeManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/MergeManager.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/MergeManager.java
index 8a09aab..dd3f73d 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/MergeManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/MergeManager.java
@@ -744,16 +744,17 @@ public class MergeManager {
ifileReadAheadLength, ifileBufferSize);
this.kvIter = kvIter;
}
- public boolean nextRawKey(DataInputBuffer key) throws IOException {
+ @Override
+ public KeyState readRawKey(DataInputBuffer key) throws IOException {
if (kvIter.next()) {
final DataInputBuffer kb = kvIter.getKey();
final int kp = kb.getPosition();
final int klen = kb.getLength() - kp;
key.reset(kb.getData(), kp, klen);
bytesRead += klen;
- return true;
+ return KeyState.NEW_KEY;
}
- return false;
+ return KeyState.NO_KEY;
}
public void nextRawValue(DataInputBuffer value) throws IOException {
final DataInputBuffer vb = kvIter.getValue();
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/2efe086d/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java
index e7545fc..e8af024 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java
@@ -506,7 +506,7 @@ public class IFile {
return true;
}
- public boolean nextRawKey(DataInputBuffer key) throws IOException {
+ public final boolean nextRawKey(DataInputBuffer key) throws IOException {
return readRawKey(key) != KeyState.NO_KEY;
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/2efe086d/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/TezMerger.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/TezMerger.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/TezMerger.java
index be21280..653b302 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/TezMerger.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/TezMerger.java
@@ -43,6 +43,7 @@ import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.counters.TezCounter;
import org.apache.tez.runtime.library.common.Constants;
import org.apache.tez.runtime.library.common.sort.impl.IFile.Reader;
+import org.apache.tez.runtime.library.common.sort.impl.IFile.Reader.KeyState;
import org.apache.tez.runtime.library.common.sort.impl.IFile.Writer;
@@ -313,6 +314,10 @@ public class TezMerger {
segmentLength : reader.getLength();
}
+ KeyState readRawKey() throws IOException {
+ return reader.readRawKey(key);
+ }
+
boolean nextRawKey() throws IOException {
return reader.nextRawKey(key);
}
@@ -472,15 +477,17 @@ public class TezMerger {
private void adjustPriorityQueue(Segment reader) throws IOException{
long startPos = reader.getPosition();
- boolean hasNext = reader.nextRawKey();
+ KeyState hasNext = reader.readRawKey();
long endPos = reader.getPosition();
totalBytesProcessed += endPos - startPos;
mergeProgress.set(totalBytesProcessed * progPerByte);
- if (hasNext) {
+ if (hasNext == KeyState.NEW_KEY) {
adjustTop();
- } else {
+ } else if(hasNext == KeyState.NO_KEY) {
pop();
reader.close();
+ } else if(hasNext == KeyState.SAME_KEY) {
+ // do not rebalance the priority queue
}
}