You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by tillrohrmann <gi...@git.apache.org> on 2014/07/01 02:16:24 UTC
[GitHub] incubator-flink pull request: [FLINK-994] Replaced DataInput and D...
GitHub user tillrohrmann opened a pull request:
https://github.com/apache/incubator-flink/pull/53
[FLINK-994] Replaced DataInput and DataOutput with DataInputView and DataOutputView
I replaced the DataInput and DataOutput parameters by DataInputView and DataOutputView in the IOReadableWritable interface. Thus, I also added wrappers for DataInputStream, DataOutputStream, ObjectOutputStream and ObjectInputStream to make them compatible with the new read and write methods.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tillrohrmann/incubator-flink FLINK-994
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-flink/pull/53.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #53
----
commit bf0d7cf252f51fc3e79e2655d91de1781940bd5b
Author: Till Rohrmann <ti...@gmail.com>
Date: 2014-06-29T17:08:31Z
Replaced DataInput and DataOutput with DataInputView and DataOutputView
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-994] Replaced DataInput and D...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/53#discussion_r14593536
--- Diff: stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/AbstractPagedInputView.java ---
@@ -183,44 +183,69 @@ protected void clear() {
// --------------------------------------------------------------------------------------------
// Data Input Specific methods
// --------------------------------------------------------------------------------------------
-
+
@Override
- public void readFully(byte[] b) throws IOException {
- readFully(b, 0, b.length);
+ public int read(byte[] b) throws IOException{
+ return read(b,0,b.length);
}
@Override
- public void readFully(byte[] b, int off, int len) throws IOException {
+ public int read(byte[] b, int off, int len) throws IOException{
if (off < 0 || len < 0 || off + len > b.length) {
throw new IndexOutOfBoundsException();
}
-
+
int remaining = this.limitInSegment - this.positionInSegment;
if (remaining >= len) {
this.currentSegment.get(this.positionInSegment, b, off, len);
this.positionInSegment += len;
+ return len;
}
else {
if (remaining == 0) {
- advance();
+ try {
+ advance();
+ }catch(EOFException eof){
+ return -1;
+ }
remaining = this.limitInSegment - this.positionInSegment;
}
-
+
+ int bytesRead = 0;
while (true) {
- int toRead = Math.min(remaining, len);
+ int toRead = Math.min(remaining, len-bytesRead);
this.currentSegment.get(this.positionInSegment, b, off, toRead);
off += toRead;
- len -= toRead;
-
- if (len > 0) {
- advance();
- remaining = this.limitInSegment - this.positionInSegment;
+ bytesRead += toRead;
+
+ if (len > bytesRead) {
+ try {
+ advance();
+ }catch(EOFException eof){
+ return bytesRead;
+ }
+ remaining = this.limitInSegment - this.positionInSegment;
}
else {
this.positionInSegment += toRead;
break;
}
}
+ return len;
+ }
+ }
+
+ @Override
+ public void readFully(byte[] b) throws IOException {
+ readFully(b, 0, b.length);
+ }
+
+ @Override
+ public void readFully(byte[] b, int off, int len) throws IOException {
+ int bytesRead = read(b,off,len);
+
+ if(bytesRead == -1){
--- End diff --
Should this be `if (bytesRead < len)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-994] Replaced DataInput and D...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/53#discussion_r14593600
--- Diff: stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/AbstractPagedInputView.java ---
@@ -183,44 +183,69 @@ protected void clear() {
// --------------------------------------------------------------------------------------------
// Data Input Specific methods
// --------------------------------------------------------------------------------------------
-
+
@Override
- public void readFully(byte[] b) throws IOException {
- readFully(b, 0, b.length);
+ public int read(byte[] b) throws IOException{
+ return read(b,0,b.length);
}
@Override
- public void readFully(byte[] b, int off, int len) throws IOException {
+ public int read(byte[] b, int off, int len) throws IOException{
if (off < 0 || len < 0 || off + len > b.length) {
throw new IndexOutOfBoundsException();
}
-
+
int remaining = this.limitInSegment - this.positionInSegment;
if (remaining >= len) {
this.currentSegment.get(this.positionInSegment, b, off, len);
this.positionInSegment += len;
+ return len;
}
else {
if (remaining == 0) {
- advance();
+ try {
+ advance();
+ }catch(EOFException eof){
+ return -1;
+ }
remaining = this.limitInSegment - this.positionInSegment;
}
-
+
+ int bytesRead = 0;
while (true) {
- int toRead = Math.min(remaining, len);
+ int toRead = Math.min(remaining, len-bytesRead);
this.currentSegment.get(this.positionInSegment, b, off, toRead);
off += toRead;
- len -= toRead;
-
- if (len > 0) {
- advance();
- remaining = this.limitInSegment - this.positionInSegment;
+ bytesRead += toRead;
+
+ if (len > bytesRead) {
+ try {
+ advance();
+ }catch(EOFException eof){
+ return bytesRead;
+ }
+ remaining = this.limitInSegment - this.positionInSegment;
}
else {
this.positionInSegment += toRead;
break;
}
}
+ return len;
+ }
+ }
+
+ @Override
+ public void readFully(byte[] b) throws IOException {
+ readFully(b, 0, b.length);
+ }
+
+ @Override
+ public void readFully(byte[] b, int off, int len) throws IOException {
+ int bytesRead = read(b,off,len);
+
+ if(bytesRead == -1){
--- End diff --
Could you add a small test for this method? Seems like complex logic...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-994] Replaced DataInput and D...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/incubator-flink/pull/53#issuecomment-48319802
LGTM. I will merge it...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-994] Replaced DataInput and D...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/53#discussion_r14594163
--- Diff: stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/AbstractPagedInputView.java ---
@@ -183,44 +183,69 @@ protected void clear() {
// --------------------------------------------------------------------------------------------
// Data Input Specific methods
// --------------------------------------------------------------------------------------------
-
+
@Override
- public void readFully(byte[] b) throws IOException {
- readFully(b, 0, b.length);
+ public int read(byte[] b) throws IOException{
+ return read(b,0,b.length);
}
@Override
- public void readFully(byte[] b, int off, int len) throws IOException {
+ public int read(byte[] b, int off, int len) throws IOException{
if (off < 0 || len < 0 || off + len > b.length) {
throw new IndexOutOfBoundsException();
}
-
+
int remaining = this.limitInSegment - this.positionInSegment;
if (remaining >= len) {
this.currentSegment.get(this.positionInSegment, b, off, len);
this.positionInSegment += len;
+ return len;
}
else {
if (remaining == 0) {
- advance();
+ try {
+ advance();
+ }catch(EOFException eof){
+ return -1;
+ }
remaining = this.limitInSegment - this.positionInSegment;
}
-
+
+ int bytesRead = 0;
while (true) {
- int toRead = Math.min(remaining, len);
+ int toRead = Math.min(remaining, len-bytesRead);
this.currentSegment.get(this.positionInSegment, b, off, toRead);
off += toRead;
- len -= toRead;
-
- if (len > 0) {
- advance();
- remaining = this.limitInSegment - this.positionInSegment;
+ bytesRead += toRead;
+
+ if (len > bytesRead) {
+ try {
+ advance();
+ }catch(EOFException eof){
+ return bytesRead;
+ }
+ remaining = this.limitInSegment - this.positionInSegment;
}
else {
this.positionInSegment += toRead;
break;
}
}
+ return len;
+ }
+ }
+
+ @Override
+ public void readFully(byte[] b) throws IOException {
+ readFully(b, 0, b.length);
+ }
+
+ @Override
+ public void readFully(byte[] b, int off, int len) throws IOException {
+ int bytesRead = read(b,off,len);
+
+ if(bytesRead == -1){
--- End diff --
Yes, you are right that's a bug. I'll add the test cases.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-994] Replaced DataInput and D...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/incubator-flink/pull/53#issuecomment-48173874
Comment inline, otherwise looks good.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-994] Replaced DataInput and D...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/incubator-flink/pull/53#issuecomment-48184382
Added the test cases and replaced DataInput/Output with DataInput/OutputView in hbase.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-994] Replaced DataInput and D...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-flink/pull/53
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---