scwhittle commented on code in PR #32398:
URL: https://github.com/apache/beam/pull/32398#discussion_r1746636153
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextSource.java:
##########
@@ -377,106 +392,61 @@ private boolean readDefaultLine() throws IOException {
return true;
}
- /**
- * Loosely based upon <a
- *
href="https://github.com/hanborq/hadoop/blob/master/src/core/org/apache/hadoop/util/LineReader.java";>Hadoop
- * LineReader.java</a>
- *
- * <p>Note that this implementation fixes an issue where a partial match
against the delimiter
- * would have been lost if the delimiter crossed at the buffer boundaries
during reading.
- */
private boolean readCustomLine() throws IOException {
- assert !eof;
+ checkState(!eof);
+ checkNotNull(delimiter);
+ checkNotNull(
+ delimiterFinder, "DelimiterFinder must not be null if custom
delimiter is used.");
long bytesConsumed = 0;
- int delPosn = 0;
- EOF:
- for (; ; ) {
- int startPosn = bufferPosn; // starting from where we left off the
last time
+ delimiterFinder.reset();
- // Read the next chunk from the file, ensure that we read at least one
byte
- // or reach EOF.
+ while (true) {
+ int startPosn = bufferPosn;
while (bufferPosn >= bufferLength) {
startPosn = bufferPosn = 0;
byteBuffer.clear();
bufferLength = inChannel.read(byteBuffer);
- // If we are at EOF then try to create the last value from the
buffer.
if (bufferLength < 0) {
eof = true;
- // Write any partial delimiter now that we are at EOF
- if (delPosn != 0) {
- str.write(delimiter, 0, delPosn);
- }
-
- // Don't return an empty record if the file ends with a delimiter
if (str.size() == 0) {
return false;
}
+ // Not ending with a delimiter.
currentValue = str.toString(StandardCharsets.UTF_8.name());
- break EOF;
+ break;
}
}
- int prevDelPosn = delPosn;
- DELIMITER_MATCH:
- {
- if (delPosn > 0) {
- // slow-path: Handle the case where we only matched part of the
delimiter, possibly
- // adding that to str fixing up any partially consumed delimiter
if we don't match the
- // whole delimiter
- for (; bufferPosn < bufferLength; ++bufferPosn) {
- if (buffer[bufferPosn] == delimiter[delPosn]) {
- delPosn++;
- if (delPosn == delimiter.length) {
- bufferPosn++;
- break DELIMITER_MATCH; // Skip matching the delimiter using
the fast path
- }
- } else {
- // Add to str any previous partial delimiter since we didn't
match the whole
- // delimiter
- str.write(delimiter, 0, prevDelPosn);
- if (buffer[bufferPosn] == delimiter[0]) {
- delPosn = 1;
- } else {
- delPosn = 0;
- }
- break; // Leave this loop and use the fast-path delimiter
matching
- }
- }
- }
+ if (eof) {
+ break;
+ }
- // fast-path: Look for the delimiter within the buffer
- for (; bufferPosn < bufferLength; ++bufferPosn) {
- if (buffer[bufferPosn] == delimiter[delPosn]) {
- delPosn++;
- if (delPosn == delimiter.length) {
- bufferPosn++;
- break;
- }
- } else if (buffer[bufferPosn] == delimiter[0]) {
- delPosn = 1;
- } else {
- delPosn = 0;
- }
+ boolean delimiterFound = false;
+ for (; bufferPosn < bufferLength; ++bufferPosn) {
Review Comment:
I think you could move startPosn = bufferPosn into loop
`for (int startPosn = bufferPosn; ...) {`
seems nice in that it initializes it as it is being consumed.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextSource.java:
##########
@@ -116,8 +118,14 @@ static class TextBasedReader extends
FileBasedReader<String> {
private final byte @Nullable [] delimiter;
private final int skipHeaderLines;
- private final ByteArrayOutputStream str;
+
+ // The output stream can contain the delimiter at the last. It must
exclude the delimiter when
Review Comment:
How about replacing the first sentence nd maybe add something about how this
is used:
"Used to build up results that span buffers. It may contain the delimiter as
a suffix."
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextSource.java:
##########
@@ -487,4 +457,110 @@ private boolean readCustomLine() throws IOException {
return true;
}
}
+
+ /**
+ * This class is created to avoid multiple bytes-copy when making a
substring of the output.
+ * Without this class, it requires two bytes copies.
+ *
+ * <pre>{@code
+ * ByteArrayOutputStream out = ...;
+ * byte[] buffer = out.toByteArray(); // 1st-copy
+ * String s = new String(buffer, offset, length); // 2nd-copy
+ * }</pre>
+ */
+ static class SubstringByteArrayOutputStream extends ByteArrayOutputStream {
+ public String toString(int offset, int length, Charset charset) {
+ if (offset < 0) {
+ throw new IllegalArgumentException("offset is negative: " + offset);
+ }
+ if (offset > count) {
+ throw new IllegalArgumentException(
+ "offset exceeds the buffer limit. offset: " + offset + ", limit: "
+ count);
+ }
+
+ if (length < 0) {
+ throw new IllegalArgumentException("length is negative: " + length);
+ }
+
+ if (offset + length > count) {
+ throw new IllegalArgumentException(
+ "offset + length exceeds the buffer limit. offset: "
+ + offset
+ + ", length: "
+ + length
+ + ", limit: "
+ + count);
+ }
+
+ return new String(buf, offset, length, charset);
+ }
+ }
+
+ /**
+ * @see <a
+ *
href="https://en.wikipedia.org/wiki/Knuth%E2%80%93Morris%E2%80%93Pratt_algorithm";>Knuth–Morris–Pratt
+ * algorithm</a>
+ */
+ static class KMPDelimiterFinder {
+ private final byte[] delimiter;
+ private final int[] table;
+ int k; // the current position in delimiter
Review Comment:
can we name the variable currentPos or delimiterOffset or something more
descriptive?
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextSource.java:
##########
@@ -377,106 +392,61 @@ private boolean readDefaultLine() throws IOException {
return true;
}
- /**
- * Loosely based upon <a
- *
href="https://github.com/hanborq/hadoop/blob/master/src/core/org/apache/hadoop/util/LineReader.java";>Hadoop
- * LineReader.java</a>
- *
- * <p>Note that this implementation fixes an issue where a partial match
against the delimiter
- * would have been lost if the delimiter crossed at the buffer boundaries
during reading.
- */
private boolean readCustomLine() throws IOException {
- assert !eof;
+ checkState(!eof);
+ checkNotNull(delimiter);
+ checkNotNull(
+ delimiterFinder, "DelimiterFinder must not be null if custom
delimiter is used.");
long bytesConsumed = 0;
- int delPosn = 0;
- EOF:
- for (; ; ) {
- int startPosn = bufferPosn; // starting from where we left off the
last time
+ delimiterFinder.reset();
- // Read the next chunk from the file, ensure that we read at least one
byte
- // or reach EOF.
+ while (true) {
+ int startPosn = bufferPosn;
while (bufferPosn >= bufferLength) {
startPosn = bufferPosn = 0;
byteBuffer.clear();
bufferLength = inChannel.read(byteBuffer);
- // If we are at EOF then try to create the last value from the
buffer.
if (bufferLength < 0) {
eof = true;
- // Write any partial delimiter now that we are at EOF
- if (delPosn != 0) {
- str.write(delimiter, 0, delPosn);
- }
-
- // Don't return an empty record if the file ends with a delimiter
if (str.size() == 0) {
return false;
}
+ // Not ending with a delimiter.
currentValue = str.toString(StandardCharsets.UTF_8.name());
- break EOF;
+ break;
}
}
- int prevDelPosn = delPosn;
- DELIMITER_MATCH:
- {
- if (delPosn > 0) {
- // slow-path: Handle the case where we only matched part of the
delimiter, possibly
- // adding that to str fixing up any partially consumed delimiter
if we don't match the
- // whole delimiter
- for (; bufferPosn < bufferLength; ++bufferPosn) {
- if (buffer[bufferPosn] == delimiter[delPosn]) {
- delPosn++;
- if (delPosn == delimiter.length) {
- bufferPosn++;
- break DELIMITER_MATCH; // Skip matching the delimiter using
the fast path
- }
- } else {
- // Add to str any previous partial delimiter since we didn't
match the whole
- // delimiter
- str.write(delimiter, 0, prevDelPosn);
- if (buffer[bufferPosn] == delimiter[0]) {
- delPosn = 1;
- } else {
- delPosn = 0;
- }
- break; // Leave this loop and use the fast-path delimiter
matching
- }
- }
- }
+ if (eof) {
Review Comment:
I think the flow could be simplified by removing the inner while which
prevents you from breaking out of the outer while. The innerwhile is just to
retry reading empty length buffers so I think that could be handled separately
like:
```
while (true) {
..
if (bufferPos >= bufferLength) {
startPosn = bufferPosn = 0;
byteBuffer.clear();
do {
bufferLength = inChannel.read(byteBuffer);
} while (bufferLength == 0);
...
if (bufferLength < 0) {
... // can break directly here
}
}
```
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextSourceTest.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class TextSourceTest {
+
+ @Test
+ public void testSubstringByteArrayOutputStreamSuccessful() throws
IOException {
+ TextSource.SubstringByteArrayOutputStream output =
+ new TextSource.SubstringByteArrayOutputStream();
+ assertEquals("", output.toString(0, 0, StandardCharsets.UTF_8));
+
+ output.write("ABC".getBytes(StandardCharsets.UTF_8));
+ assertEquals("ABC", output.toString(0, 3, StandardCharsets.UTF_8));
+ assertEquals("AB", output.toString(0, 2, StandardCharsets.UTF_8));
+ assertEquals("BC", output.toString(1, 2, StandardCharsets.UTF_8));
+ assertEquals("", output.toString(3, 0, StandardCharsets.UTF_8));
+
+ output.write("DE".getBytes(StandardCharsets.UTF_8));
+ assertEquals("ABCDE", output.toString(0, 5, StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testSubstringByteArrayIllegalArgumentException() throws
IOException {
+ TextSource.SubstringByteArrayOutputStream output =
+ new TextSource.SubstringByteArrayOutputStream();
+ output.write("ABC".getBytes(StandardCharsets.UTF_8));
+
+ IllegalArgumentException exception;
+ exception =
+ assertThrows(
+ IllegalArgumentException.class, () -> output.toString(-1, 2,
StandardCharsets.UTF_8));
+ assertThat(exception.getMessage(), containsString("offset is negative"));
+
+ exception =
+ assertThrows(
+ IllegalArgumentException.class, () -> output.toString(4, 0,
StandardCharsets.UTF_8));
+ assertThat(exception.getMessage(), containsString("offset exceeds the
buffer limit"));
+
+ exception =
+ assertThrows(
+ IllegalArgumentException.class, () -> output.toString(0, -1,
StandardCharsets.UTF_8));
+ assertThat(exception.getMessage(), containsString("length is negative"));
+
+ exception =
+ assertThrows(
+ IllegalArgumentException.class, () -> output.toString(2, 2,
StandardCharsets.UTF_8));
+ assertThat(exception.getMessage(), containsString("offset + length exceeds
the buffer limit"));
+ }
+
+ @Test
+ public void testDelimiterFinder() {
+ assertEquals(Arrays.asList("A", "C"), split("AB", "AABC"));
+ assertEquals(Arrays.asList("A", "B", "C"), split("AB", "AABBABC"));
+
+ assertEquals(Arrays.asList("A", "C"), split("AAB", "AAABC"));
+ assertEquals(Arrays.asList("ABAB"), split("AAB", "ABAB"));
+ assertEquals(Arrays.asList("A", "A", "B"), split("AAB", "AAABAAABB"));
+
+ assertEquals(Arrays.asList("A", "B"), split("AABA", "AAABAB"));
+ assertEquals(Arrays.asList("AABBA"), split("AABA", "AABBA"));
+
+ assertEquals(Arrays.asList("", "D"), split("ABABC", "ABABCD"));
+ assertEquals(Arrays.asList("ABABAD"), split("ABABC", "ABABAD"));
+ assertEquals(Arrays.asList("ABABBD"), split("ABABC", "ABABBD"));
+ assertEquals(Arrays.asList("AB"), split("ABABC", "ABABABC"));
+
+ assertEquals(Arrays.asList(""), split("ABCABD", "ABCABD"));
Review Comment:
I don't see a test where there are several delimiters in a row, do we want
multiple empty strings as result in that case?
ie
split("AA", "AAAAAA") -> ["", "", ""] ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
