Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1266#discussion_r44434150
--- Diff:
flink-java/src/main/java/org/apache/flink/api/java/io/CsvInputFormat.java ---
@@ -18,32 +18,97 @@
package org.apache.flink.api.java.io;
+import com.google.common.base.Preconditions;
+import com.google.common.primitives.Ints;
+import org.apache.flink.api.common.io.GenericCsvInputFormat;
+import org.apache.flink.core.fs.FileInputSplit;
+import org.apache.flink.types.parser.FieldParser;
-import org.apache.flink.api.common.typeutils.CompositeType;
-import org.apache.flink.api.java.tuple.Tuple;
+import java.io.IOException;
import org.apache.flink.core.fs.Path;
import org.apache.flink.util.StringUtils;
-public class CsvInputFormat<OUT> extends CommonCsvInputFormat<OUT> {
+public abstract class CsvInputFormat<OUT> extends
GenericCsvInputFormat<OUT> {
private static final long serialVersionUID = 1L;
+
+ public static final String DEFAULT_LINE_DELIMITER = "\n";
+
+ public static final String DEFAULT_FIELD_DELIMITER = ",";
+
+ protected transient Object[] parsedValues;
- public CsvInputFormat(Path filePath, CompositeType<OUT>
typeInformation) {
- super(filePath, typeInformation);
+ protected CsvInputFormat(Path filePath) {
+ super(filePath);
}
-
- public CsvInputFormat(Path filePath, String lineDelimiter, String
fieldDelimiter, CompositeType<OUT> typeInformation) {
- super(filePath, lineDelimiter, fieldDelimiter, typeInformation);
+
+ @Override
+ public void open(FileInputSplit split) throws IOException {
+ super.open(split);
+
+ @SuppressWarnings("unchecked")
+ FieldParser<Object>[] fieldParsers = (FieldParser<Object>[])
getFieldParsers();
+
+ //throw exception if no field parsers are available
+ if (fieldParsers.length == 0) {
+ throw new
IOException("CsvInputFormat.open(FileInputSplit split) - no field parsers to
parse input");
+ }
+
+ // create the value holders
+ this.parsedValues = new Object[fieldParsers.length];
+ for (int i = 0; i < fieldParsers.length; i++) {
+ this.parsedValues[i] = fieldParsers[i].createValue();
+ }
+
+ // left to right evaluation makes access [0] okay
+ // this marker is used to fasten up readRecord, so that it
doesn't have to check each call if the line ending is set to default
+ if (this.getDelimiter().length == 1 && this.getDelimiter()[0]
== '\n' ) {
+ this.lineDelimiterIsLinebreak = true;
+ }
+
+ this.commentCount = 0;
+ this.invalidLineCount = 0;
}
@Override
- protected OUT createTuple(OUT reuse) {
- Tuple result = (Tuple) reuse;
- for (int i = 0; i < parsedValues.length; i++) {
- result.setField(parsedValues[i], i);
+ public OUT nextRecord(OUT record) throws IOException {
+ OUT returnRecord = null;
+ do {
+ returnRecord = super.nextRecord(record);
+ } while (returnRecord == null && !reachedEnd());
+
+ return returnRecord;
+ }
+
+ public Class<?>[] getFieldTypes() {
+ return super.getGenericFieldTypes();
+ }
+
+ protected static boolean[] createDefaultMask(int size) {
--- End diff --
Isn't the default that fields are read one after the other from the start
of a line?
Why do we need this method then?
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
