You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2015/11/18 21:42:44 UTC
[1/2] flink git commit: [FLINK-2692] Untangle CsvInputFormat
Repository: flink
Updated Branches:
refs/heads/master fc6fec786 -> bd61f2dbd
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/test/java/org/apache/flink/api/java/io/CsvInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/io/CsvInputFormatTest.java b/flink-java/src/test/java/org/apache/flink/api/java/io/CsvInputFormatTest.java
index 2c2ffa5..b89f6f5 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/io/CsvInputFormatTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/io/CsvInputFormatTest.java
@@ -73,7 +73,7 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<String, Integer, Double>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class, Double.class);
- final CsvInputFormat<Tuple3<String, Integer, Double>> format = new CsvInputFormat<Tuple3<String, Integer, Double>>(PATH, "\n", "|", typeInfo);
+ final CsvInputFormat<Tuple3<String, Integer, Double>> format = new TupleCsvInputFormat<Tuple3<String, Integer, Double>>(PATH, "\n", "|", typeInfo);
format.setLenient(true);
final Configuration parameters = new Configuration();
@@ -121,7 +121,7 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<String, Integer, Double>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class, Double.class);
- final CsvInputFormat<Tuple3<String, Integer, Double>> format = new CsvInputFormat<Tuple3<String, Integer, Double>>(PATH, "\n", "|", typeInfo);
+ final CsvInputFormat<Tuple3<String, Integer, Double>> format = new TupleCsvInputFormat<Tuple3<String, Integer, Double>>(PATH, "\n", "|", typeInfo);
format.setCommentPrefix("#");
final Configuration parameters = new Configuration();
@@ -165,7 +165,7 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<String, Integer, Double>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class, Double.class);
- final CsvInputFormat<Tuple3<String, Integer, Double>> format = new CsvInputFormat<Tuple3<String, Integer, Double>>(PATH, "\n", "|", typeInfo);
+ final CsvInputFormat<Tuple3<String, Integer, Double>> format = new TupleCsvInputFormat<Tuple3<String, Integer, Double>>(PATH, "\n", "|", typeInfo);
format.setCommentPrefix("//");
final Configuration parameters = new Configuration();
@@ -202,7 +202,7 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<String, String, String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class, String.class);
- final CsvInputFormat<Tuple3<String, String, String>> format = new CsvInputFormat<Tuple3<String, String, String>>(PATH, "\n", "|", typeInfo);
+ final CsvInputFormat<Tuple3<String, String, String>> format = new TupleCsvInputFormat<Tuple3<String, String, String>>(PATH, "\n", "|", typeInfo);
final Configuration parameters = new Configuration();
format.configure(parameters);
@@ -245,7 +245,7 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<String, String, String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class, String.class);
- final CsvInputFormat<Tuple3<String, String, String>> format = new CsvInputFormat<Tuple3<String, String, String>>(PATH, "\n", "|", typeInfo);
+ final CsvInputFormat<Tuple3<String, String, String>> format = new TupleCsvInputFormat<Tuple3<String, String, String>>(PATH, "\n", "|", typeInfo);
final Configuration parameters = new Configuration();
format.configure(parameters);
@@ -289,7 +289,7 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<String, String, String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class, String.class);
- final CsvInputFormat<Tuple3<String, String, String>> format = new CsvInputFormat<Tuple3<String, String, String>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple3<String, String, String>> format = new TupleCsvInputFormat<Tuple3<String, String, String>>(PATH, typeInfo);
format.setFieldDelimiter("|-");
@@ -333,7 +333,7 @@ public class CsvInputFormatTest {
final TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>> typeInfo =
TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class, Integer.class, Integer.class, Integer.class);
- final CsvInputFormat<Tuple5<Integer, Integer, Integer, Integer, Integer>> format = new CsvInputFormat<Tuple5<Integer, Integer, Integer, Integer, Integer>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple5<Integer, Integer, Integer, Integer, Integer>> format = new TupleCsvInputFormat<Tuple5<Integer, Integer, Integer, Integer, Integer>>(PATH, typeInfo);
format.setFieldDelimiter("|");
@@ -380,7 +380,7 @@ public class CsvInputFormatTest {
final TupleTypeInfo<Tuple6<Short, Integer, Long, Float, Double, Byte>> typeInfo =
TupleTypeInfo.getBasicTupleTypeInfo(Short.class, Integer.class, Long.class, Float.class, Double.class, Byte.class);
- final CsvInputFormat<Tuple6<Short, Integer, Long, Float, Double, Byte>> format = new CsvInputFormat<Tuple6<Short, Integer, Long, Float, Double, Byte>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple6<Short, Integer, Long, Float, Double, Byte>> format = new TupleCsvInputFormat<Tuple6<Short, Integer, Long, Float, Double, Byte>>(PATH, typeInfo);
format.setFieldDelimiter("|");
@@ -431,7 +431,7 @@ public class CsvInputFormatTest {
final TupleTypeInfo<Tuple5<Double, Double, Double, Double, Double>> typeInfo =
TupleTypeInfo.getBasicTupleTypeInfo(Double.class, Double.class, Double.class, Double.class, Double.class);
- final CsvInputFormat<Tuple5<Double, Double, Double, Double, Double>> format = new CsvInputFormat<Tuple5<Double, Double, Double, Double, Double>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple5<Double, Double, Double, Double, Double>> format = new TupleCsvInputFormat<Tuple5<Double, Double, Double, Double, Double>>(PATH, typeInfo);
format.setFieldDelimiter("|");
@@ -472,7 +472,7 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple2<Integer, Integer>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class);
- final CsvInputFormat<Tuple2<Integer, Integer>> format = new CsvInputFormat<Tuple2<Integer, Integer>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple2<Integer, Integer>> format = new TupleCsvInputFormat<Tuple2<Integer, Integer>>(PATH, typeInfo);
format.setFieldDelimiter("|");
@@ -509,10 +509,9 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<Integer, Integer, Integer>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class, Integer.class);
- final CsvInputFormat<Tuple3<Integer, Integer, Integer>> format = new CsvInputFormat<Tuple3<Integer, Integer, Integer>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple3<Integer, Integer, Integer>> format = new TupleCsvInputFormat<Tuple3<Integer, Integer, Integer>>(PATH, typeInfo, new boolean[]{true, false, false, true, false, false, false, true});
format.setFieldDelimiter("|x|");
- format.setFieldTypes(Integer.class, null, null, Integer.class, null, null, null, Integer.class);
format.configure(new Configuration());
format.open(split);
@@ -547,13 +546,10 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<Integer, Integer, Integer>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class, Integer.class);
- final CsvInputFormat<Tuple3<Integer, Integer, Integer>> format = new CsvInputFormat<Tuple3<Integer, Integer, Integer>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple3<Integer, Integer, Integer>> format = new TupleCsvInputFormat<Tuple3<Integer, Integer, Integer>>(PATH, typeInfo, new int[]{0, 3, 7});
format.setFieldDelimiter("|");
- format.setFields(new int[]{0, 3, 7}, new Class<?>[]{Integer.class, Integer.class, Integer.class});
-
-
format.configure(new Configuration());
format.open(split);
@@ -579,7 +575,7 @@ public class CsvInputFormatTest {
fail("Test failed due to a " + ex.getClass().getName() + ": " + ex.getMessage());
}
}
-
+
@Test
public void testReadSparseWithMask() throws IOException {
try {
@@ -588,13 +584,9 @@ public class CsvInputFormatTest {
final FileInputSplit split = createTempFile(fileContent);
final TupleTypeInfo<Tuple3<Integer, Integer, Integer>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class, Integer.class);
- final CsvInputFormat<Tuple3<Integer, Integer, Integer>> format = new CsvInputFormat<Tuple3<Integer, Integer, Integer>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple3<Integer, Integer, Integer>> format = new TupleCsvInputFormat<Tuple3<Integer, Integer, Integer>>(PATH, typeInfo, new boolean[]{true, false, false, true, false, false, false, true});
format.setFieldDelimiter("&&");
-
- format.setFields(new boolean[]{true, false, false, true, false, false, false, true}, new
- Class<?>[]{Integer.class,
- Integer.class, Integer.class});
format.configure(new Configuration());
format.open(split);
@@ -621,27 +613,6 @@ public class CsvInputFormatTest {
fail("Test failed due to a " + ex.getClass().getName() + ": " + ex.getMessage());
}
}
-
- @Test
- public void testReadSparseWithShuffledPositions() throws IOException {
- try {
- final TupleTypeInfo<Tuple3<Integer, Integer, Integer>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class, Integer.class);
- final CsvInputFormat<Tuple3<Integer, Integer, Integer>> format = new CsvInputFormat<Tuple3<Integer, Integer, Integer>>(PATH, typeInfo);
-
- format.setFieldDelimiter("|");
-
- try {
- format.setFields(new int[] {8, 1, 3}, new Class<?>[] {Integer.class, Integer.class, Integer.class});
- fail("Input sequence should have been rejected.");
- }
- catch (IllegalArgumentException e) {
- // that is what we want
- }
- }
- catch (Exception ex) {
- fail("Test failed due to a " + ex.getClass().getName() + ": " + ex.getMessage());
- }
- }
@Test
public void testParseStringErrors() throws Exception {
@@ -682,14 +653,11 @@ public class CsvInputFormatTest {
final TupleTypeInfo<Tuple5<Integer, String, String, String, Double>> typeInfo =
TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, String.class, String.class, String.class, Double.class);
- final CsvInputFormat<Tuple5<Integer, String, String, String, Double>> format = new CsvInputFormat<Tuple5<Integer, String, String, String, Double>>(PATH, typeInfo);
+ final CsvInputFormat<Tuple5<Integer, String, String, String, Double>> format = new TupleCsvInputFormat<Tuple5<Integer, String, String, String, Double>>(PATH, typeInfo);
format.setSkipFirstLineAsHeader(true);
format.setFieldDelimiter(',');
- format.setFields(new boolean[] { true, true, true, true, true }, new Class<?>[] {
- Integer.class, String.class, String.class, String.class, Double.class });
-
format.configure(new Configuration());
format.open(split);
@@ -764,7 +732,7 @@ public class CsvInputFormatTest {
wrt.close();
final TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
- final CsvInputFormat<Tuple1<String>> inputFormat = new CsvInputFormat<Tuple1<String>>(new Path(tempFile.toURI().toString()), typeInfo);
+ final CsvInputFormat<Tuple1<String>> inputFormat = new TupleCsvInputFormat<Tuple1<String>>(new Path(tempFile.toURI().toString()), typeInfo);
Configuration parameters = new Configuration();
inputFormat.configure(parameters);
@@ -827,7 +795,7 @@ public class CsvInputFormatTest {
@SuppressWarnings("unchecked")
PojoTypeInfo<PojoItem> typeInfo = (PojoTypeInfo<PojoItem>) TypeExtractor.createTypeInfo(PojoItem.class);
- CsvInputFormat<PojoItem> inputFormat = new CsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo);
+ CsvInputFormat<PojoItem> inputFormat = new PojoCsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo);
inputFormat.configure(new Configuration());
FileInputSplit[] splits = inputFormat.createInputSplits(1);
@@ -850,7 +818,7 @@ public class CsvInputFormatTest {
@SuppressWarnings("unchecked")
PojoTypeInfo<PrivatePojoItem> typeInfo = (PojoTypeInfo<PrivatePojoItem>) TypeExtractor.createTypeInfo(PrivatePojoItem.class);
- CsvInputFormat<PrivatePojoItem> inputFormat = new CsvInputFormat<PrivatePojoItem>(new Path(tempFile.toURI().toString()), typeInfo);
+ CsvInputFormat<PrivatePojoItem> inputFormat = new PojoCsvInputFormat<PrivatePojoItem>(new Path(tempFile.toURI().toString()), typeInfo);
inputFormat.configure(new Configuration());
@@ -886,9 +854,7 @@ public class CsvInputFormatTest {
@SuppressWarnings("unchecked")
PojoTypeInfo<PojoItem> typeInfo = (PojoTypeInfo<PojoItem>) TypeExtractor.createTypeInfo(PojoItem.class);
- CsvInputFormat<PojoItem> inputFormat = new CsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo);
- inputFormat.setFields(new boolean[]{true, true, true, true}, new Class<?>[]{Integer.class, Double.class, String.class, String.class});
- inputFormat.setOrderOfPOJOFields(new String[]{"field1", "field3", "field2", "field4"});
+ CsvInputFormat<PojoItem> inputFormat = new PojoCsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo, new String[]{"field1", "field3", "field2", "field4"});
inputFormat.configure(new Configuration());
FileInputSplit[] splits = inputFormat.createInputSplits(1);
@@ -911,9 +877,7 @@ public class CsvInputFormatTest {
@SuppressWarnings("unchecked")
PojoTypeInfo<PojoItem> typeInfo = (PojoTypeInfo<PojoItem>) TypeExtractor.createTypeInfo(PojoItem.class);
- CsvInputFormat<PojoItem> inputFormat = new CsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo);
- inputFormat.setFields(new boolean[]{true, false, true, false, true, true}, new Class[]{Integer.class, String
- .class, Double.class, String.class});
+ CsvInputFormat<PojoItem> inputFormat = new PojoCsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo, new boolean[]{true, false, true, false, true, true});
inputFormat.configure(new Configuration());
FileInputSplit[] splits = inputFormat.createInputSplits(1);
@@ -936,9 +900,7 @@ public class CsvInputFormatTest {
@SuppressWarnings("unchecked")
PojoTypeInfo<PojoItem> typeInfo = (PojoTypeInfo<PojoItem>) TypeExtractor.createTypeInfo(PojoItem.class);
- CsvInputFormat<PojoItem> inputFormat = new CsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo);
- inputFormat.setFields(new boolean[]{true, false, false, true}, new Class[]{Integer.class, String.class});
- inputFormat.setOrderOfPOJOFields(new String[]{"field1", "field4"});
+ CsvInputFormat<PojoItem> inputFormat = new PojoCsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo, new String[]{"field1", "field4"}, new boolean[]{true, false, false, true});
inputFormat.configure(new Configuration());
FileInputSplit[] splits = inputFormat.createInputSplits(1);
@@ -960,24 +922,23 @@ public class CsvInputFormatTest {
@SuppressWarnings("unchecked")
PojoTypeInfo<PojoItem> typeInfo = (PojoTypeInfo<PojoItem>) TypeExtractor.createTypeInfo(PojoItem.class);
- CsvInputFormat<PojoItem> inputFormat = new CsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo);
try {
- inputFormat.setOrderOfPOJOFields(new String[]{"field1", "field2"});
+ new PojoCsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo, new String[]{"field1", "field2"});
fail("The number of POJO fields cannot be same as that of selected CSV fields");
} catch (IllegalArgumentException e) {
// success
}
try {
- inputFormat.setOrderOfPOJOFields(new String[]{"field1", "field2", null, "field4"});
+ new PojoCsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo, new String[]{"field1", "field2", null, "field4"});
fail("Fields mapping cannot contain null.");
} catch (NullPointerException e) {
// success
}
try {
- inputFormat.setOrderOfPOJOFields(new String[]{"field1", "field2", "field3", "field5"});
+ new PojoCsvInputFormat<PojoItem>(new Path(tempFile.toURI().toString()), typeInfo, new String[]{"field1", "field2", "field3", "field5"});
fail("Invalid field name");
} catch (IllegalArgumentException e) {
// success
@@ -997,13 +958,12 @@ public class CsvInputFormatTest {
writer.write(fileContent);
writer.close();
- CompositeType<Tuple2<String, String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class);
- CsvInputFormat<Tuple2<String, String>> inputFormat = new CsvInputFormat<Tuple2<String, String>>(new Path(tempFile.toURI().toString()), typeInfo);
+ TupleTypeInfo<Tuple2<String, String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class);
+ CsvInputFormat<Tuple2<String, String>> inputFormat = new TupleCsvInputFormat<Tuple2<String, String>>(new Path(tempFile.toURI().toString()), typeInfo, new boolean[]{true, false, true});
inputFormat.enableQuotedStringParsing('"');
inputFormat.setFieldDelimiter('|');
inputFormat.setDelimiter('\n');
- inputFormat.setFields(new boolean[]{true, false, true}, new Class[]{String.class, String.class});
inputFormat.configure(new Configuration());
FileInputSplit[] splits = inputFormat.createInputSplits(1);
@@ -1029,7 +989,7 @@ public class CsvInputFormatTest {
writer.close();
TupleTypeInfo<Tuple2<String, String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class);
- CsvInputFormat<Tuple2<String, String>> inputFormat = new CsvInputFormat<>(new Path(tempFile.toURI().toString()), typeInfo);
+ CsvInputFormat<Tuple2<String, String>> inputFormat = new TupleCsvInputFormat<>(new Path(tempFile.toURI().toString()), typeInfo);
inputFormat.enableQuotedStringParsing('"');
inputFormat.setFieldDelimiter('|');
@@ -1064,7 +1024,7 @@ public class CsvInputFormatTest {
@SuppressWarnings("unchecked")
PojoTypeInfo<TwitterPOJO> typeInfo = (PojoTypeInfo<TwitterPOJO>)TypeExtractor.createTypeInfo(TwitterPOJO.class);
- CsvInputFormat<TwitterPOJO> inputFormat = new CsvInputFormat<>(new Path(tempFile.toURI().toString()), typeInfo);
+ CsvInputFormat<TwitterPOJO> inputFormat = new PojoCsvInputFormat<>(new Path(tempFile.toURI().toString()), typeInfo);
inputFormat.configure(new Configuration());
FileInputSplit[] splits = inputFormat.createInputSplits(1);
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
index 6c0a375..f4f501a 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
@@ -17,12 +17,11 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.HashMap;
-import org.apache.flink.api.common.typeutils.CompositeType;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.LocalEnvironment;
-import org.apache.flink.api.java.io.CsvInputFormat;
import org.apache.flink.api.java.io.PrintingOutputFormat;
+import org.apache.flink.api.java.io.TupleCsvInputFormat;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.CoGroupRawOperator;
import org.apache.flink.api.java.operators.CrossOperator.DefaultCross;
@@ -35,6 +34,7 @@ import org.apache.flink.api.java.operators.SortedGrouping;
import org.apache.flink.api.java.operators.UdfOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.core.fs.FileSystem;
@@ -435,13 +435,12 @@ public class PythonPlanBinder {
}
private void createCsvSource(PythonOperationInfo info) throws IOException {
- if (!(info.types instanceof CompositeType)) {
- throw new RuntimeException("The output type of a csv source has to be a tuple or a "
- + "pojo type. The derived type is " + info);
+ if (!(info.types instanceof TupleTypeInfo)) {
+ throw new RuntimeException("The output type of a csv source has to be a tuple. The derived type is " + info);
}
- sets.put(info.setID, env.createInput(new CsvInputFormat(new Path(info.path),
- info.lineDelimiter, info.fieldDelimiter, (CompositeType) info.types), info.types)
+ sets.put(info.setID, env.createInput(new TupleCsvInputFormat(new Path(info.path),
+ info.lineDelimiter, info.fieldDelimiter, (TupleTypeInfo) info.types), info.types)
.name("CsvSource"));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-optimizer/src/test/java/org/apache/flink/optimizer/ReplicatingDataSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/ReplicatingDataSourceTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/ReplicatingDataSourceTest.java
index 26af380..cb01436 100644
--- a/flink-optimizer/src/test/java/org/apache/flink/optimizer/ReplicatingDataSourceTest.java
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/ReplicatingDataSourceTest.java
@@ -27,7 +27,7 @@ import org.apache.flink.api.common.io.ReplicatingInputFormat;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.io.CsvInputFormat;
+import org.apache.flink.api.java.io.TupleCsvInputFormat;
import org.apache.flink.api.java.operators.DataSink;
import org.apache.flink.api.java.operators.translation.JavaPlan;
import org.apache.flink.api.java.tuple.Tuple1;
@@ -58,7 +58,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -95,7 +95,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -133,7 +133,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -171,7 +171,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -209,7 +209,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -247,7 +247,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -288,7 +288,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -325,7 +325,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -364,7 +364,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -390,7 +390,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -417,7 +417,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
@@ -443,7 +443,7 @@ public class ReplicatingDataSourceTest extends CompilerTestBase {
TupleTypeInfo<Tuple1<String>> typeInfo = TupleTypeInfo.getBasicTupleTypeInfo(String.class);
ReplicatingInputFormat<Tuple1<String>, FileInputSplit> rif =
- new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new CsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
+ new ReplicatingInputFormat<Tuple1<String>, FileInputSplit>(new TupleCsvInputFormat<Tuple1<String>>(new Path("/some/path"), typeInfo));
DataSet<Tuple1<String>> source1 = env.createInput(rif, new TupleTypeInfo<Tuple1<String>>(BasicTypeInfo.STRING_TYPE_INFO));
DataSet<Tuple1<String>> source2 = env.readCsvFile("/some/otherpath").types(String.class);
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvInputFormat.java b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvInputFormat.java
deleted file mode 100644
index 522cc0c..0000000
--- a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvInputFormat.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.flink.api.scala.operators;
-
-
-import com.google.common.base.Preconditions;
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.CompositeType;
-import org.apache.flink.api.java.io.CommonCsvInputFormat;
-import org.apache.flink.api.java.typeutils.PojoTypeInfo;
-import org.apache.flink.api.java.typeutils.TupleTypeInfoBase;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase;
-import org.apache.flink.core.fs.Path;
-import org.apache.flink.util.StringUtils;
-
-public class ScalaCsvInputFormat<OUT> extends CommonCsvInputFormat<OUT> {
- private static final long serialVersionUID = -7347888812778968640L;
-
- private final TupleSerializerBase<OUT> tupleSerializer;
-
- public ScalaCsvInputFormat(Path filePath, CompositeType<OUT> typeInfo) {
- super(filePath, typeInfo);
-
- Preconditions.checkArgument(typeInfo instanceof PojoTypeInfo || typeInfo instanceof TupleTypeInfoBase,
- "Only pojo types or tuple types are supported.");
-
- if (typeInfo instanceof TupleTypeInfoBase) {
- TupleTypeInfoBase<OUT> tupleTypeInfo = (TupleTypeInfoBase<OUT>) typeInfo;
-
- tupleSerializer = (TupleSerializerBase<OUT>)tupleTypeInfo.createSerializer(new ExecutionConfig());
- } else {
- tupleSerializer = null;
- }
- }
-
- @Override
- protected OUT createTuple(OUT reuse) {
- Preconditions.checkNotNull(tupleSerializer, "The tuple serializer must be initialised." +
- " It is not initialized if the given type was not a " +
- TupleTypeInfoBase.class.getName() + ".");
-
- return tupleSerializer.createInstance(parsedValues);
- }
-
- @Override
- public String toString() {
- return "Scala CSV Input (" + StringUtils.showControlCharacters(String.valueOf(getFieldDelimiter())) + ") " + getFilePath();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java
index e29a6e6..afcdc4a 100644
--- a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java
+++ b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java
@@ -23,7 +23,7 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.io.FileOutputFormat;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.io.CommonCsvInputFormat;
+import org.apache.flink.api.java.io.CsvInputFormat;
import org.apache.flink.api.java.typeutils.InputTypeConfigurable;
import org.apache.flink.core.fs.Path;
import org.apache.flink.types.StringValue;
@@ -51,9 +51,9 @@ public class ScalaCsvOutputFormat<T extends Product> extends FileOutputFormat<T>
// --------------------------------------------------------------------------------------------
- public static final String DEFAULT_LINE_DELIMITER = CommonCsvInputFormat.DEFAULT_LINE_DELIMITER;
+ public static final String DEFAULT_LINE_DELIMITER = CsvInputFormat.DEFAULT_LINE_DELIMITER;
- public static final String DEFAULT_FIELD_DELIMITER = String.valueOf(CommonCsvInputFormat.DEFAULT_FIELD_DELIMITER);
+ public static final String DEFAULT_FIELD_DELIMITER = String.valueOf(CsvInputFormat.DEFAULT_FIELD_DELIMITER);
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
index 6381a12..cb576dd 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
@@ -29,7 +29,6 @@ import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfoBase, ValueTypeInfo}
import org.apache.flink.api.java.{CollectionEnvironment, ExecutionEnvironment => JavaEnv}
import org.apache.flink.api.scala.hadoop.{mapred, mapreduce}
-import org.apache.flink.api.scala.operators.ScalaCsvInputFormat
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.Path
import org.apache.flink.types.StringValue
@@ -279,55 +278,34 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
s"The type $typeInfo has to be a tuple or pojo type.",
null)
- val inputFormat = new ScalaCsvInputFormat[T](
- new Path(filePath),
- typeInfo.asInstanceOf[CompositeType[T]])
- inputFormat.setDelimiter(lineDelimiter)
- inputFormat.setFieldDelimiter(fieldDelimiter)
- inputFormat.setSkipFirstLineAsHeader(ignoreFirstLine)
- inputFormat.setLenient(lenient)
- inputFormat.setCommentPrefix(ignoreComments)
+ var inputFormat: CsvInputFormat[T] = null
- if (quoteCharacter != null) {
- inputFormat.enableQuotedStringParsing(quoteCharacter)
- }
-
- val classesBuf: ArrayBuffer[Class[_]] = new ArrayBuffer[Class[_]]
typeInfo match {
case info: TupleTypeInfoBase[T] =>
- for (i <- 0 until info.getArity) {
- classesBuf += info.getTypeAt(i).getTypeClass()
- }
+ inputFormat = new TupleCsvInputFormat[T](
+ new Path(filePath),
+ typeInfo.asInstanceOf[TupleTypeInfoBase[T]],
+ includedFields)
case info: PojoTypeInfo[T] =>
if (pojoFields == null) {
throw new IllegalArgumentException(
"POJO fields must be specified (not null) if output type is a POJO.")
- } else {
- for (i <- pojoFields.indices) {
- val pos = info.getFieldIndex(pojoFields(i))
- if (pos < 0) {
- throw new IllegalArgumentException(
- "Field \"" + pojoFields(i) + "\" not part of POJO type " +
- info.getTypeClass.getCanonicalName)
- }
- classesBuf += info.getPojoFieldAt(pos).getTypeInformation().getTypeClass
- }
}
+ inputFormat = new PojoCsvInputFormat[T](
+ new Path(filePath),
+ typeInfo.asInstanceOf[PojoTypeInfo[T]],
+ pojoFields,
+ includedFields)
case _ => throw new IllegalArgumentException("Type information is not valid.")
}
-
- if (includedFields != null) {
- require(classesBuf.size == includedFields.length, "Number of tuple fields and" +
- " included fields must match.")
- inputFormat.setFields(includedFields, classesBuf.toArray)
- } else {
- inputFormat.setFieldTypes(classesBuf: _*)
- }
-
- if (pojoFields != null) {
- inputFormat.setOrderOfPOJOFields(pojoFields)
+ if (quoteCharacter != null) {
+ inputFormat.enableQuotedStringParsing(quoteCharacter)
}
-
+ inputFormat.setDelimiter(lineDelimiter)
+ inputFormat.setFieldDelimiter(fieldDelimiter)
+ inputFormat.setSkipFirstLineAsHeader(ignoreFirstLine)
+ inputFormat.setLenient(lenient)
+ inputFormat.setCommentPrefix(ignoreComments)
wrap(new DataSource[T](javaEnv, inputFormat, typeInfo, getCallLocationName()))
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassSerializer.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassSerializer.scala
index 3a015f7..3824f9a 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassSerializer.scala
@@ -73,6 +73,10 @@ abstract class CaseClassSerializer[T <: Product](
}
}
+ override def createOrReuseInstance(fields: Array[Object], reuse: T) : T = {
+ createInstance(fields)
+ }
+
def copy(from: T, reuse: T): T = {
copy(from)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-tests/src/test/scala/org/apache/flink/api/scala/io/CsvInputFormatTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/io/CsvInputFormatTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/io/CsvInputFormatTest.scala
index e49f737..47f99ca 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/io/CsvInputFormatTest.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/io/CsvInputFormatTest.scala
@@ -21,10 +21,11 @@ import java.io.{File, FileOutputStream, FileWriter, OutputStreamWriter}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.io.PojoCsvInputFormat
+import org.apache.flink.api.java.io.TupleCsvInputFormat
import org.apache.flink.api.java.io.CsvInputFormatTest.TwitterPOJO
-import org.apache.flink.api.java.typeutils.{TupleTypeInfo, PojoTypeInfo}
+import org.apache.flink.api.java.typeutils.PojoTypeInfo
import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.operators.ScalaCsvInputFormat
import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.{FileInputSplit, Path}
@@ -50,7 +51,7 @@ class CsvInputFormatTest {
"a test|3|4.0|\n" +
"#next|5|6.0|\n"
val split = createTempFile(fileContent)
- val format = new ScalaCsvInputFormat[(String, Integer, Double)](
+ val format = new TupleCsvInputFormat[(String, Integer, Double)](
PATH,
createTypeInformation[(String, Integer, Double)]
.asInstanceOf[CaseClassTypeInfo[(String, Integer, Double)]])
@@ -92,7 +93,7 @@ class CsvInputFormatTest {
"a test|3|4.0|\n" +
"//next|5|6.0|\n"
val split = createTempFile(fileContent)
- val format = new ScalaCsvInputFormat[(String, Integer, Double)](
+ val format = new TupleCsvInputFormat[(String, Integer, Double)](
PATH,
createTypeInformation[(String, Integer, Double)]
.asInstanceOf[CaseClassTypeInfo[(String, Integer, Double)]])
@@ -130,7 +131,7 @@ class CsvInputFormatTest {
try {
val fileContent = "abc|def|ghijk\nabc||hhg\n|||"
val split = createTempFile(fileContent)
- val format = new ScalaCsvInputFormat[(String, String, String)](
+ val format = new TupleCsvInputFormat[(String, String, String)](
PATH,
createTypeInformation[(String, String, String)]
.asInstanceOf[CaseClassTypeInfo[(String, String, String)]])
@@ -172,7 +173,7 @@ class CsvInputFormatTest {
try {
val fileContent = "abc|\"de|f\"|ghijk\n\"a|bc\"||hhg\n|||"
val split = createTempFile(fileContent)
- val format = new ScalaCsvInputFormat[(String, String, String)](
+ val format = new TupleCsvInputFormat[(String, String, String)](
PATH,
createTypeInformation[(String, String, String)]
.asInstanceOf[CaseClassTypeInfo[(String, String, String)]])
@@ -215,7 +216,7 @@ class CsvInputFormatTest {
try {
val fileContent = "abc|-def|-ghijk\nabc|-|-hhg\n|-|-|-\n"
val split = createTempFile(fileContent)
- val format = new ScalaCsvInputFormat[(String, String, String)](
+ val format = new TupleCsvInputFormat[(String, String, String)](
PATH,
createTypeInformation[(String, String, String)]
.asInstanceOf[CaseClassTypeInfo[(String, String, String)]])
@@ -256,7 +257,7 @@ class CsvInputFormatTest {
try {
val fileContent = "111|222|333|444|555\n666|777|888|999|000|\n"
val split = createTempFile(fileContent)
- val format = new ScalaCsvInputFormat[(Int, Int, Int, Int, Int)](
+ val format = new TupleCsvInputFormat[(Int, Int, Int, Int, Int)](
PATH, createTypeInformation[(Int, Int, Int, Int, Int)].
asInstanceOf[CaseClassTypeInfo[(Int, Int, Int, Int, Int)]])
format.setFieldDelimiter("|")
@@ -293,7 +294,7 @@ class CsvInputFormatTest {
val fileContent = "111|x|222|x|333|x|444|x|555|x|\n" +
"666|x|777|x|888|x|999|x|000|x|\n"
val split = createTempFile(fileContent)
- val format = new ScalaCsvInputFormat[(Int, Int)](
+ val format = new TupleCsvInputFormat[(Int, Int)](
PATH,
createTypeInformation[(Int, Int)].asInstanceOf[CaseClassTypeInfo[(Int, Int)]])
format.setFieldDelimiter("|x|")
@@ -317,19 +318,17 @@ class CsvInputFormatTest {
fail("Test failed due to a " + ex.getClass.getName + ": " + ex.getMessage)
}
}
-
@Test
def testReadSparseWithPositionSetter(): Unit = {
try {
val fileContent: String = "111|222|333|444|555|666|777|888|999|000|\n000|999|888|777|666" +
"|555|444|333|222|111|"
val split = createTempFile(fileContent)
- val format = new ScalaCsvInputFormat[(Int, Int, Int)](
+ val format = new TupleCsvInputFormat[(Int, Int, Int)](
PATH,
- createTypeInformation[(Int, Int, Int)].asInstanceOf[CaseClassTypeInfo[(Int, Int, Int)]])
+ createTypeInformation[(Int, Int, Int)].asInstanceOf[CaseClassTypeInfo[(Int, Int, Int)]],
+ Array(0, 3, 7))
format.setFieldDelimiter("|")
- format.setFields(Array(0, 3, 7),
- Array(classOf[Integer], classOf[Integer], classOf[Integer]): Array[Class[_]])
format.configure(new Configuration)
format.open(split)
var result: (Int, Int, Int) = null
@@ -353,28 +352,6 @@ class CsvInputFormatTest {
}
}
- @Test
- def testReadSparseWithShuffledPositions(): Unit = {
- try {
- val format = new ScalaCsvInputFormat[(Int, Int, Int)](
- PATH,
- createTypeInformation[(Int, Int, Int)].asInstanceOf[CaseClassTypeInfo[(Int, Int, Int)]])
- format.setFieldDelimiter("|")
- try {
- format.setFields(Array(8, 1, 3),
- Array(classOf[Integer], classOf[Integer], classOf[Integer]): Array[Class[_]])
- fail("Input sequence should have been rejected.")
- }
- catch {
- case e: IllegalArgumentException => // ignore
- }
- }
- catch {
- case ex: Exception =>
- fail("Test failed due to a " + ex.getClass.getName + ": " + ex.getMessage)
- }
- }
-
private def createTempFile(content: String): FileInputSplit = {
val tempFile = File.createTempFile("test_contents", "tmp")
tempFile.deleteOnExit()
@@ -402,7 +379,7 @@ class CsvInputFormatTest {
val wrt = new OutputStreamWriter(new FileOutputStream(tempFile))
wrt.write(fileContent)
wrt.close()
- val inputFormat = new ScalaCsvInputFormat[Tuple1[String]](new Path(tempFile.toURI.toString),
+ val inputFormat = new TupleCsvInputFormat[Tuple1[String]](new Path(tempFile.toURI.toString),
createTypeInformation[Tuple1[String]].asInstanceOf[CaseClassTypeInfo[Tuple1[String]]])
val parameters = new Configuration
inputFormat.configure(parameters)
@@ -432,7 +409,7 @@ class CsvInputFormatTest {
case class CaseClassItem(field1: Int, field2: String, field3: Double)
- private def validatePOJOItem(format: ScalaCsvInputFormat[POJOItem]): Unit = {
+ private def validatePOJOItem(format: PojoCsvInputFormat[POJOItem]): Unit = {
var result = new POJOItem()
result = format.nextRecord(result)
assertEquals(123, result.field1)
@@ -445,7 +422,7 @@ class CsvInputFormatTest {
assertEquals(1.234, result.field3, 0.001)
}
- private def validateCaseClassItem(format: ScalaCsvInputFormat[CaseClassItem]): Unit = {
+ private def validateCaseClassItem(format: TupleCsvInputFormat[CaseClassItem]): Unit = {
var result = format.nextRecord(null)
assertEquals(123, result.field1)
assertEquals("HELLO", result.field2)
@@ -463,7 +440,7 @@ class CsvInputFormatTest {
val tempFile = createTempFile(fileContent)
val typeInfo: PojoTypeInfo[POJOItem] = createTypeInformation[POJOItem]
.asInstanceOf[PojoTypeInfo[POJOItem]]
- val format = new ScalaCsvInputFormat[POJOItem](PATH, typeInfo)
+ val format = new PojoCsvInputFormat[POJOItem](PATH, typeInfo)
format.setDelimiter('\n')
format.setFieldDelimiter(',')
@@ -477,9 +454,10 @@ class CsvInputFormatTest {
def testCaseClass(): Unit = {
val fileContent = "123,HELLO,3.123\n" + "456,ABC,1.234"
val tempFile = createTempFile(fileContent)
- val typeInfo: CompositeType[CaseClassItem] = createTypeInformation[CaseClassItem]
- .asInstanceOf[CompositeType[CaseClassItem]]
- val format = new ScalaCsvInputFormat[CaseClassItem](PATH, typeInfo)
+ val typeInfo: CaseClassTypeInfo[CaseClassItem] =
+ createTypeInformation[CaseClassItem]
+ .asInstanceOf[CaseClassTypeInfo[CaseClassItem]]
+ val format = new TupleCsvInputFormat[CaseClassItem](PATH, typeInfo)
format.setDelimiter('\n')
format.setFieldDelimiter(',')
@@ -495,12 +473,11 @@ class CsvInputFormatTest {
val tempFile = createTempFile(fileContent)
val typeInfo: PojoTypeInfo[POJOItem] = createTypeInformation[POJOItem]
.asInstanceOf[PojoTypeInfo[POJOItem]]
- val format = new ScalaCsvInputFormat[POJOItem](PATH, typeInfo)
+ val format = new PojoCsvInputFormat[POJOItem](
+ PATH, typeInfo, Array("field2", "field1", "field3"))
format.setDelimiter('\n')
format.setFieldDelimiter(',')
- format.setFieldTypes(classOf[String], classOf[Integer], classOf[java.lang.Double])
- format.setOrderOfPOJOFields(Array("field2", "field1", "field3"))
format.configure(new Configuration)
format.open(tempFile)
@@ -513,13 +490,12 @@ class CsvInputFormatTest {
val tempFile = createTempFile(fileContent)
val typeInfo: PojoTypeInfo[POJOItem] = createTypeInformation[POJOItem]
.asInstanceOf[PojoTypeInfo[POJOItem]]
- val format = new ScalaCsvInputFormat[POJOItem](PATH, typeInfo)
+ val format = new PojoCsvInputFormat[POJOItem](
+ PATH, typeInfo, Array("field2", "field1", "field3"),
+ Array(true, true, false, true, false))
format.setDelimiter('\n')
format.setFieldDelimiter(',')
- format.setFields(Array(true, true, false, true, false),
- Array(classOf[String], classOf[Integer], classOf[java.lang.Double]): Array[Class[_]])
- format.setOrderOfPOJOFields(Array("field2", "field1", "field3"))
format.configure(new Configuration)
format.open(tempFile)
@@ -532,7 +508,7 @@ class CsvInputFormatTest {
val tempFile = createTempFile(fileContent)
val typeInfo: PojoTypeInfo[TwitterPOJO] = createTypeInformation[TwitterPOJO]
.asInstanceOf[PojoTypeInfo[TwitterPOJO]]
- val format = new ScalaCsvInputFormat[TwitterPOJO](PATH, typeInfo)
+ val format = new PojoCsvInputFormat[TwitterPOJO](PATH, typeInfo)
format.setDelimiter('\n')
format.setFieldDelimiter(',')
[2/2] flink git commit: [FLINK-2692] Untangle CsvInputFormat
Posted by ch...@apache.org.
[FLINK-2692] Untangle CsvInputFormat
This closes #1266
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bd61f2db
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bd61f2db
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bd61f2db
Branch: refs/heads/master
Commit: bd61f2dbdf1a0215363ffa8416329e1dbf277593
Parents: fc6fec7
Author: zentol <ch...@apache.org>
Authored: Sun Oct 18 20:23:23 2015 +0200
Committer: zentol <ch...@apache.org>
Committed: Wed Nov 18 21:42:00 2015 +0100
----------------------------------------------------------------------
.../wordcount/BoltTokenizerWordCountPojo.java | 3 +-
.../BoltTokenizerWordCountWithNames.java | 3 +-
.../flink/api/java/io/CommonCsvInputFormat.java | 258 -------------------
.../flink/api/java/io/CsvInputFormat.java | 127 ++++++++-
.../org/apache/flink/api/java/io/CsvReader.java | 125 ++++-----
.../flink/api/java/io/PojoCsvInputFormat.java | 199 ++++++++++++++
.../flink/api/java/io/TupleCsvInputFormat.java | 90 +++++++
.../flink/api/java/tuple/TupleGenerator.java | 13 +-
.../java/typeutils/runtime/TupleSerializer.java | 8 +
.../typeutils/runtime/TupleSerializerBase.java | 2 +
.../flink/api/java/io/CsvInputFormatTest.java | 96 ++-----
.../flink/python/api/PythonPlanBinder.java | 13 +-
.../optimizer/ReplicatingDataSourceTest.java | 26 +-
.../scala/operators/ScalaCsvInputFormat.java | 65 -----
.../scala/operators/ScalaCsvOutputFormat.java | 6 +-
.../flink/api/scala/ExecutionEnvironment.scala | 56 ++--
.../scala/typeutils/CaseClassSerializer.scala | 4 +
.../flink/api/scala/io/CsvInputFormatTest.scala | 78 ++----
18 files changed, 573 insertions(+), 599 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
index 9bdcead..e093714 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
@@ -20,6 +20,7 @@ package org.apache.flink.storm.wordcount;
import backtype.storm.topology.IRichBolt;
import org.apache.flink.api.java.io.CsvInputFormat;
+import org.apache.flink.api.java.io.PojoCsvInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
@@ -119,7 +120,7 @@ public class BoltTokenizerWordCountPojo {
// read the text file from given input path
PojoTypeInfo<Sentence> sourceType = (PojoTypeInfo<Sentence>) TypeExtractor
.getForObject(new Sentence(""));
- return env.createInput(new CsvInputFormat<Sentence>(new Path(
+ return env.createInput(new PojoCsvInputFormat<Sentence>(new Path(
textPath), CsvInputFormat.DEFAULT_LINE_DELIMITER,
CsvInputFormat.DEFAULT_LINE_DELIMITER, sourceType),
sourceType);
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
index 019f1bc..f5a1a35 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
@@ -21,6 +21,7 @@ import backtype.storm.topology.IRichBolt;
import backtype.storm.tuple.Fields;
import org.apache.flink.api.java.io.CsvInputFormat;
+import org.apache.flink.api.java.io.TupleCsvInputFormat;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -122,7 +123,7 @@ public class BoltTokenizerWordCountWithNames {
// read the text file from given input path
TupleTypeInfo<Tuple1<String>> sourceType = (TupleTypeInfo<Tuple1<String>>)TypeExtractor
.getForObject(new Tuple1<String>(""));
- return env.createInput(new CsvInputFormat<Tuple1<String>>(new Path(
+ return env.createInput(new TupleCsvInputFormat<Tuple1<String>>(new Path(
textPath), CsvInputFormat.DEFAULT_LINE_DELIMITER,
CsvInputFormat.DEFAULT_LINE_DELIMITER, sourceType),
sourceType);
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/main/java/org/apache/flink/api/java/io/CommonCsvInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/CommonCsvInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/io/CommonCsvInputFormat.java
deleted file mode 100644
index 444d151..0000000
--- a/flink-java/src/main/java/org/apache/flink/api/java/io/CommonCsvInputFormat.java
+++ /dev/null
@@ -1,258 +0,0 @@
-/*
- * 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.flink.api.java.io;
-
-import com.google.common.base.Preconditions;
-import org.apache.flink.api.common.io.GenericCsvInputFormat;
-import org.apache.flink.api.common.typeutils.CompositeType;
-import org.apache.flink.api.java.typeutils.PojoTypeInfo;
-import org.apache.flink.core.fs.FileInputSplit;
-import org.apache.flink.core.fs.Path;
-import org.apache.flink.types.parser.FieldParser;
-
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-
-public abstract class CommonCsvInputFormat<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;
-
- private final Class<OUT> pojoTypeClass;
-
- private String[] pojoFieldNames;
-
- private transient PojoTypeInfo<OUT> pojoTypeInfo;
- private transient Field[] pojoFields;
-
- public CommonCsvInputFormat(Path filePath, CompositeType<OUT> typeInformation) {
- this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, typeInformation);
- }
-
- public CommonCsvInputFormat(
- Path filePath,
- String lineDelimiter,
- String fieldDelimiter,
- CompositeType<OUT> compositeTypeInfo) {
- super(filePath);
-
- setDelimiter(lineDelimiter);
- setFieldDelimiter(fieldDelimiter);
-
- Class<?>[] classes = new Class<?>[compositeTypeInfo.getArity()];
-
- for (int i = 0; i < compositeTypeInfo.getArity(); i++) {
- classes[i] = compositeTypeInfo.getTypeAt(i).getTypeClass();
- }
-
- setFieldTypes(classes);
-
- if (compositeTypeInfo instanceof PojoTypeInfo) {
- pojoTypeInfo = (PojoTypeInfo<OUT>) compositeTypeInfo;
-
- pojoTypeClass = compositeTypeInfo.getTypeClass();
- setOrderOfPOJOFields(compositeTypeInfo.getFieldNames());
- } else {
- pojoTypeClass = null;
- pojoFieldNames = null;
- }
- }
-
- public void setOrderOfPOJOFields(String[] fieldNames) {
- Preconditions.checkNotNull(pojoTypeClass, "Field order can only be specified if output type is a POJO.");
- Preconditions.checkNotNull(fieldNames);
-
- int includedCount = 0;
- for (boolean isIncluded : fieldIncluded) {
- if (isIncluded) {
- includedCount++;
- }
- }
-
- Preconditions.checkArgument(includedCount == fieldNames.length, includedCount +
- " CSV fields and " + fieldNames.length + " POJO fields selected. The number of selected CSV and POJO fields must be equal.");
-
- for (String field : fieldNames) {
- Preconditions.checkNotNull(field, "The field name cannot be null.");
- Preconditions.checkArgument(pojoTypeInfo.getFieldIndex(field) != -1,
- "Field \""+ field + "\" is not a member of POJO class " + pojoTypeClass.getName());
- }
-
- pojoFieldNames = Arrays.copyOfRange(fieldNames, 0, fieldNames.length);
- }
-
- public void setFieldTypes(Class<?>... fieldTypes) {
- if (fieldTypes == null || fieldTypes.length == 0) {
- throw new IllegalArgumentException("Field types must not be null or empty.");
- }
-
- setFieldTypesGeneric(fieldTypes);
- }
-
- public void setFields(int[] sourceFieldIndices, Class<?>[] fieldTypes) {
- Preconditions.checkNotNull(sourceFieldIndices);
- Preconditions.checkNotNull(fieldTypes);
-
- checkForMonotonousOrder(sourceFieldIndices, fieldTypes);
-
- setFieldsGeneric(sourceFieldIndices, fieldTypes);
- }
-
- public void setFields(boolean[] sourceFieldMask, Class<?>[] fieldTypes) {
- Preconditions.checkNotNull(sourceFieldMask);
- Preconditions.checkNotNull(fieldTypes);
-
- setFieldsGeneric(sourceFieldMask, fieldTypes);
- }
-
- public Class<?>[] getFieldTypes() {
- return super.getGenericFieldTypes();
- }
-
- @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;
- }
-
- // for POJO type
- if (pojoTypeClass != null) {
- pojoFields = new Field[pojoFieldNames.length];
-
- Map<String, Field> allFields = new HashMap<String, Field>();
-
- findAllFields(pojoTypeClass, allFields);
-
- for (int i = 0; i < pojoFieldNames.length; i++) {
- pojoFields[i] = allFields.get(pojoFieldNames[i]);
-
- if (pojoFields[i] != null) {
- pojoFields[i].setAccessible(true);
- } else {
- throw new RuntimeException("There is no field called \"" + pojoFieldNames[i] + "\" in " + pojoTypeClass.getName());
- }
- }
- }
-
- this.commentCount = 0;
- this.invalidLineCount = 0;
- }
-
- /**
- * Finds all declared fields in a class and all its super classes.
- *
- * @param clazz Class for which all declared fields are found
- * @param allFields Map containing all found fields so far
- */
- private void findAllFields(Class<?> clazz, Map<String, Field> allFields) {
- for (Field field: clazz.getDeclaredFields()) {
- allFields.put(field.getName(), field);
- }
-
- if (clazz.getSuperclass() != null) {
- findAllFields(clazz.getSuperclass(), allFields);
- }
- }
-
- @Override
- public OUT nextRecord(OUT record) throws IOException {
- OUT returnRecord = null;
- do {
- returnRecord = super.nextRecord(record);
- } while (returnRecord == null && !reachedEnd());
-
- return returnRecord;
- }
-
- @Override
- public OUT readRecord(OUT reuse, byte[] bytes, int offset, int numBytes) throws IOException {
- /*
- * Fix to support windows line endings in CSVInputFiles with standard delimiter setup = \n
- */
- //Find windows end line, so find carriage return before the newline
- if (this.lineDelimiterIsLinebreak == true && numBytes > 0 && bytes[offset + numBytes -1] == '\r' ) {
- //reduce the number of bytes so that the Carriage return is not taken as data
- numBytes--;
- }
-
- if (commentPrefix != null && commentPrefix.length <= numBytes) {
- //check record for comments
- boolean isComment = true;
- for (int i = 0; i < commentPrefix.length; i++) {
- if (commentPrefix[i] != bytes[offset + i]) {
- isComment = false;
- break;
- }
- }
- if (isComment) {
- this.commentCount++;
- return null;
- }
- }
-
- if (parseRecord(parsedValues, bytes, offset, numBytes)) {
- if (pojoTypeClass == null) {
- // result type is tuple
- return createTuple(reuse);
- } else {
- // result type is POJO
- for (int i = 0; i < parsedValues.length; i++) {
- try {
- pojoFields[i].set(reuse, parsedValues[i]);
- } catch (IllegalAccessException e) {
- throw new RuntimeException("Parsed value could not be set in POJO field \"" + pojoFieldNames[i] + "\"", e);
- }
- }
- return reuse;
- }
-
- } else {
- this.invalidLineCount++;
- return null;
- }
- }
-
- protected abstract OUT createTuple(OUT reuse);
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/main/java/org/apache/flink/api/java/io/CsvInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/CsvInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/io/CsvInputFormat.java
index 7d86f39..8f0aa64 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/io/CsvInputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/io/CsvInputFormat.java
@@ -18,32 +18,133 @@
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
+ public OUT nextRecord(OUT record) throws IOException {
+ OUT returnRecord = null;
+ do {
+ returnRecord = super.nextRecord(record);
+ } while (returnRecord == null && !reachedEnd());
+
+ return returnRecord;
}
@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 readRecord(OUT reuse, byte[] bytes, int offset, int numBytes) throws IOException {
+ /*
+ * Fix to support windows line endings in CSVInputFiles with standard delimiter setup = \n
+ */
+ //Find windows end line, so find carriage return before the newline
+ if (this.lineDelimiterIsLinebreak == true && numBytes > 0 && bytes[offset + numBytes - 1] == '\r') {
+ //reduce the number of bytes so that the Carriage return is not taken as data
+ numBytes--;
+ }
+
+ if (commentPrefix != null && commentPrefix.length <= numBytes) {
+ //check record for comments
+ boolean isComment = true;
+ for (int i = 0; i < commentPrefix.length; i++) {
+ if (commentPrefix[i] != bytes[offset + i]) {
+ isComment = false;
+ break;
+ }
+ }
+ if (isComment) {
+ this.commentCount++;
+ return null;
+ }
+ }
+
+ if (parseRecord(parsedValues, bytes, offset, numBytes)) {
+ return fillRecord(reuse, parsedValues);
+ } else {
+ this.invalidLineCount++;
+ return null;
+ }
+ }
+
+ protected abstract OUT fillRecord(OUT reuse, Object[] parsedValues);
+
+ public Class<?>[] getFieldTypes() {
+ return super.getGenericFieldTypes();
+ }
+
+ protected static boolean[] createDefaultMask(int size) {
+ boolean[] includedMask = new boolean[size];
+ for (int x=0; x<includedMask.length; x++) {
+ includedMask[x] = true;
+ }
+ return includedMask;
+ }
+
+ protected static boolean[] toBooleanMask(int[] sourceFieldIndices) {
+ Preconditions.checkNotNull(sourceFieldIndices);
+
+ for (int i : sourceFieldIndices) {
+ if (i < 0) {
+ throw new IllegalArgumentException("Field indices must not be smaller than zero.");
+ }
+ }
+
+ boolean[] includedMask = new boolean[Ints.max(sourceFieldIndices) + 1];
+
+ // check if we support parsers for these types
+ for (int i = 0; i < sourceFieldIndices.length; i++) {
+ includedMask[sourceFieldIndices[i]] = true;
}
- return reuse;
+ return includedMask;
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java b/flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java
index 34b5b47..1bb2eb9 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java
@@ -292,19 +292,10 @@ public class CsvReader {
@SuppressWarnings("unchecked")
PojoTypeInfo<T> typeInfo = (PojoTypeInfo<T>) TypeExtractor.createTypeInfo(pojoType);
- CsvInputFormat<T> inputFormat = new CsvInputFormat<T>(path, typeInfo);
- Class<?>[] classes = new Class<?>[pojoFields.length];
- for (int i = 0; i < pojoFields.length; i++) {
- int pos = typeInfo.getFieldIndex(pojoFields[i]);
- if(pos < 0) {
- throw new IllegalArgumentException("Field \""+pojoFields[i]+"\" not part of POJO type "+pojoType.getCanonicalName());
- }
- classes[i] = typeInfo.getPojoFieldAt(pos).getTypeInformation().getTypeClass();
- }
+ CsvInputFormat<T> inputFormat = new PojoCsvInputFormat<T>(path, this.lineDelimiter, this.fieldDelimiter, typeInfo, pojoFields, this.includedMask);
- configureInputFormat(inputFormat, classes);
- inputFormat.setOrderOfPOJOFields(pojoFields);
+ configureInputFormat(inputFormat);
return new DataSource<T>(executionContext, inputFormat, typeInfo, Utils.getCallLocationName());
}
@@ -325,14 +316,14 @@ public class CsvReader {
@SuppressWarnings("unchecked")
TupleTypeInfo<T> typeInfo = (TupleTypeInfo<T>) TypeExtractor.createTypeInfo(targetType);
- CsvInputFormat<T> inputFormat = new CsvInputFormat<T>(path, typeInfo);
+ CsvInputFormat<T> inputFormat = new TupleCsvInputFormat<T>(path, this.lineDelimiter, this.fieldDelimiter, typeInfo, this.includedMask);
Class<?>[] classes = new Class<?>[typeInfo.getArity()];
for (int i = 0; i < typeInfo.getArity(); i++) {
classes[i] = typeInfo.getTypeAt(i).getTypeClass();
}
- configureInputFormat(inputFormat, classes);
+ configureInputFormat(inputFormat);
return new DataSource<T>(executionContext, inputFormat, typeInfo, Utils.getCallLocationName());
}
@@ -340,7 +331,7 @@ public class CsvReader {
// Miscellaneous
// --------------------------------------------------------------------------------------------
- private void configureInputFormat(CsvInputFormat<?> format, Class<?>... types) {
+ private void configureInputFormat(CsvInputFormat<?> format) {
format.setDelimiter(this.lineDelimiter);
format.setFieldDelimiter(this.fieldDelimiter);
format.setCommentPrefix(this.commentPrefix);
@@ -349,12 +340,6 @@ public class CsvReader {
if (this.parseQuotedStrings) {
format.enableQuotedStringParsing(this.quoteCharacter);
}
-
- if (this.includedMask == null) {
- format.setFieldTypes(types);
- } else {
- format.setFields(this.includedMask, types);
- }
}
// --------------------------------------------------------------------------------------------
@@ -374,8 +359,8 @@ public class CsvReader {
*/
public <T0> DataSource<Tuple1<T0>> types(Class<T0> type0) {
TupleTypeInfo<Tuple1<T0>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0);
- CsvInputFormat<Tuple1<T0>> inputFormat = new CsvInputFormat<Tuple1<T0>>(path, types);
- configureInputFormat(inputFormat, type0);
+ CsvInputFormat<Tuple1<T0>> inputFormat = new TupleCsvInputFormat<Tuple1<T0>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple1<T0>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -391,8 +376,8 @@ public class CsvReader {
*/
public <T0, T1> DataSource<Tuple2<T0, T1>> types(Class<T0> type0, Class<T1> type1) {
TupleTypeInfo<Tuple2<T0, T1>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1);
- CsvInputFormat<Tuple2<T0, T1>> inputFormat = new CsvInputFormat<Tuple2<T0, T1>>(path, types);
- configureInputFormat(inputFormat, type0, type1);
+ CsvInputFormat<Tuple2<T0, T1>> inputFormat = new TupleCsvInputFormat<Tuple2<T0, T1>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple2<T0, T1>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -409,8 +394,8 @@ public class CsvReader {
*/
public <T0, T1, T2> DataSource<Tuple3<T0, T1, T2>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2) {
TupleTypeInfo<Tuple3<T0, T1, T2>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2);
- CsvInputFormat<Tuple3<T0, T1, T2>> inputFormat = new CsvInputFormat<Tuple3<T0, T1, T2>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2);
+ CsvInputFormat<Tuple3<T0, T1, T2>> inputFormat = new TupleCsvInputFormat<Tuple3<T0, T1, T2>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple3<T0, T1, T2>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -428,8 +413,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3> DataSource<Tuple4<T0, T1, T2, T3>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3) {
TupleTypeInfo<Tuple4<T0, T1, T2, T3>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3);
- CsvInputFormat<Tuple4<T0, T1, T2, T3>> inputFormat = new CsvInputFormat<Tuple4<T0, T1, T2, T3>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3);
+ CsvInputFormat<Tuple4<T0, T1, T2, T3>> inputFormat = new TupleCsvInputFormat<Tuple4<T0, T1, T2, T3>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple4<T0, T1, T2, T3>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -448,8 +433,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4> DataSource<Tuple5<T0, T1, T2, T3, T4>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4) {
TupleTypeInfo<Tuple5<T0, T1, T2, T3, T4>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4);
- CsvInputFormat<Tuple5<T0, T1, T2, T3, T4>> inputFormat = new CsvInputFormat<Tuple5<T0, T1, T2, T3, T4>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4);
+ CsvInputFormat<Tuple5<T0, T1, T2, T3, T4>> inputFormat = new TupleCsvInputFormat<Tuple5<T0, T1, T2, T3, T4>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple5<T0, T1, T2, T3, T4>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -469,8 +454,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5> DataSource<Tuple6<T0, T1, T2, T3, T4, T5>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5) {
TupleTypeInfo<Tuple6<T0, T1, T2, T3, T4, T5>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5);
- CsvInputFormat<Tuple6<T0, T1, T2, T3, T4, T5>> inputFormat = new CsvInputFormat<Tuple6<T0, T1, T2, T3, T4, T5>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5);
+ CsvInputFormat<Tuple6<T0, T1, T2, T3, T4, T5>> inputFormat = new TupleCsvInputFormat<Tuple6<T0, T1, T2, T3, T4, T5>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple6<T0, T1, T2, T3, T4, T5>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -491,8 +476,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6> DataSource<Tuple7<T0, T1, T2, T3, T4, T5, T6>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6) {
TupleTypeInfo<Tuple7<T0, T1, T2, T3, T4, T5, T6>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6);
- CsvInputFormat<Tuple7<T0, T1, T2, T3, T4, T5, T6>> inputFormat = new CsvInputFormat<Tuple7<T0, T1, T2, T3, T4, T5, T6>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6);
+ CsvInputFormat<Tuple7<T0, T1, T2, T3, T4, T5, T6>> inputFormat = new TupleCsvInputFormat<Tuple7<T0, T1, T2, T3, T4, T5, T6>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple7<T0, T1, T2, T3, T4, T5, T6>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -514,8 +499,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7> DataSource<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7) {
TupleTypeInfo<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7);
- CsvInputFormat<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>> inputFormat = new CsvInputFormat<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7);
+ CsvInputFormat<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>> inputFormat = new TupleCsvInputFormat<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -538,8 +523,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8> DataSource<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8) {
TupleTypeInfo<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8);
- CsvInputFormat<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>> inputFormat = new CsvInputFormat<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8);
+ CsvInputFormat<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>> inputFormat = new TupleCsvInputFormat<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -563,8 +548,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9> DataSource<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9) {
TupleTypeInfo<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9);
- CsvInputFormat<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>> inputFormat = new CsvInputFormat<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9);
+ CsvInputFormat<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>> inputFormat = new TupleCsvInputFormat<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -589,8 +574,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10> DataSource<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10) {
TupleTypeInfo<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10);
- CsvInputFormat<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>> inputFormat = new CsvInputFormat<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10);
+ CsvInputFormat<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>> inputFormat = new TupleCsvInputFormat<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -616,8 +601,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11> DataSource<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11) {
TupleTypeInfo<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11);
- CsvInputFormat<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>> inputFormat = new CsvInputFormat<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11);
+ CsvInputFormat<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>> inputFormat = new TupleCsvInputFormat<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -644,8 +629,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12> DataSource<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12) {
TupleTypeInfo<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12);
- CsvInputFormat<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>> inputFormat = new CsvInputFormat<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12);
+ CsvInputFormat<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>> inputFormat = new TupleCsvInputFormat<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -673,8 +658,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13> DataSource<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13) {
TupleTypeInfo<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13);
- CsvInputFormat<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>> inputFormat = new CsvInputFormat<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13);
+ CsvInputFormat<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>> inputFormat = new TupleCsvInputFormat<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -703,8 +688,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14> DataSource<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14) {
TupleTypeInfo<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14);
- CsvInputFormat<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>> inputFormat = new CsvInputFormat<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14);
+ CsvInputFormat<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>> inputFormat = new TupleCsvInputFormat<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -734,8 +719,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15> DataSource<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15) {
TupleTypeInfo<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15);
- CsvInputFormat<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>> inputFormat = new CsvInputFormat<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15);
+ CsvInputFormat<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>> inputFormat = new TupleCsvInputFormat<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -766,8 +751,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16> DataSource<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16) {
TupleTypeInfo<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16);
- CsvInputFormat<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>> inputFormat = new CsvInputFormat<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16);
+ CsvInputFormat<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>> inputFormat = new TupleCsvInputFormat<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -799,8 +784,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17> DataSource<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16, Class<T17> type17) {
TupleTypeInfo<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17);
- CsvInputFormat<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>> inputFormat = new CsvInputFormat<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17);
+ CsvInputFormat<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>> inputFormat = new TupleCsvInputFormat<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -833,8 +818,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18> DataSource<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16, Class<T17> type17, Class<T18> type18) {
TupleTypeInfo<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18);
- CsvInputFormat<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>> inputFormat = new CsvInputFormat<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18);
+ CsvInputFormat<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>> inputFormat = new TupleCsvInputFormat<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -868,8 +853,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19> DataSource<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16, Class<T17> type17, Class<T18> type18, Class<T19> type19) {
TupleTypeInfo<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19);
- CsvInputFormat<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>> inputFormat = new CsvInputFormat<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19);
+ CsvInputFormat<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>> inputFormat = new TupleCsvInputFormat<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -904,8 +889,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20> DataSource<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16, Class<T17> type17, Class<T18> type18, Class<T19> type19, Class<T20> type20) {
TupleTypeInfo<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20);
- CsvInputFormat<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>> inputFormat = new CsvInputFormat<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20);
+ CsvInputFormat<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>> inputFormat = new TupleCsvInputFormat<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -941,8 +926,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21> DataSource<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16, Class<T17> type17, Class<T18> type18, Class<T19> type19, Class<T20> type20, Class<T21> type21) {
TupleTypeInfo<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20, type21);
- CsvInputFormat<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>> inputFormat = new CsvInputFormat<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20, type21);
+ CsvInputFormat<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>> inputFormat = new TupleCsvInputFormat<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -979,8 +964,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22> DataSource<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16, Class<T17> type17, Class<T18> type18, Class<T19> type19, Class<T20> type20, Class<T21> type21, Class<T22> type22) {
TupleTypeInfo<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20, type21, type22);
- CsvInputFormat<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>> inputFormat = new CsvInputFormat<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20, type21, type22);
+ CsvInputFormat<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>> inputFormat = new TupleCsvInputFormat<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -1018,8 +1003,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23> DataSource<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16, Class<T17> type17, Class<T18> type18, Class<T19> type19, Class<T20> type20, Class<T21> type21, Class<T22> type22, Class<T23> type23) {
TupleTypeInfo<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20, type21, type22, type23);
- CsvInputFormat<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>> inputFormat = new CsvInputFormat<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20, type21, type22, type23);
+ CsvInputFormat<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>> inputFormat = new TupleCsvInputFormat<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
@@ -1058,8 +1043,8 @@ public class CsvReader {
*/
public <T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24> DataSource<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>> types(Class<T0> type0, Class<T1> type1, Class<T2> type2, Class<T3> type3, Class<T4> type4, Class<T5> type5, Class<T6> type6, Class<T7> type7, Class<T8> type8, Class<T9> type9, Class<T10> type10, Class<T11> type11, Class<T12> type12, Class<T13> type13, Class<T14> type14, Class<T15> type15, Class<T16> type16, Class<T17> type17, Class<T18> type18, Class<T19> type19, Class<T20> type20, Class<T21> type21, Class<T22> type22, Class<T23> type23, Class<T24> type24) {
TupleTypeInfo<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>> types = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20, type21, type22, type23, type24);
- CsvInputFormat<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>> inputFormat = new CsvInputFormat<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>>(path, types);
- configureInputFormat(inputFormat, type0, type1, type2, type3, type4, type5, type6, type7, type8, type9, type10, type11, type12, type13, type14, type15, type16, type17, type18, type19, type20, type21, type22, type23, type24);
+ CsvInputFormat<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>> inputFormat = new TupleCsvInputFormat<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>>(path, types, this.includedMask);
+ configureInputFormat(inputFormat);
return new DataSource<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>>(executionContext, inputFormat, types, Utils.getCallLocationName());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/main/java/org/apache/flink/api/java/io/PojoCsvInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/PojoCsvInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/io/PojoCsvInputFormat.java
new file mode 100644
index 0000000..2f1139c
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/io/PojoCsvInputFormat.java
@@ -0,0 +1,199 @@
+/*
+ * 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.flink.api.java.io;
+
+import com.google.common.base.Preconditions;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.core.fs.FileInputSplit;
+import org.apache.flink.core.fs.Path;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+public class PojoCsvInputFormat<OUT> extends CsvInputFormat<OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ private Class<OUT> pojoTypeClass;
+
+ private String[] pojoFieldNames;
+
+ private transient PojoTypeInfo<OUT> pojoTypeInfo;
+ private transient Field[] pojoFields;
+
+ public PojoCsvInputFormat(Path filePath, PojoTypeInfo<OUT> pojoTypeInfo) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, pojoTypeInfo);
+ }
+
+ public PojoCsvInputFormat(Path filePath, PojoTypeInfo<OUT> pojoTypeInfo, String[] fieldNames) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, pojoTypeInfo, fieldNames, createDefaultMask(pojoTypeInfo.getArity()));
+ }
+
+ public PojoCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, PojoTypeInfo<OUT> pojoTypeInfo) {
+ this(filePath, lineDelimiter, fieldDelimiter, pojoTypeInfo, pojoTypeInfo.getFieldNames(), createDefaultMask(pojoTypeInfo.getArity()));
+ }
+
+ public PojoCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, PojoTypeInfo<OUT> pojoTypeInfo, String[] fieldNames) {
+ this(filePath, lineDelimiter, fieldDelimiter, pojoTypeInfo, fieldNames, createDefaultMask(fieldNames.length));
+ }
+
+ public PojoCsvInputFormat(Path filePath, PojoTypeInfo<OUT> pojoTypeInfo, int[] includedFieldsMask) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, pojoTypeInfo, pojoTypeInfo.getFieldNames(), toBooleanMask(includedFieldsMask));
+ }
+
+ public PojoCsvInputFormat(Path filePath, PojoTypeInfo<OUT> pojoTypeInfo, String[] fieldNames, int[] includedFieldsMask) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, pojoTypeInfo, fieldNames, includedFieldsMask);
+ }
+
+ public PojoCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, PojoTypeInfo<OUT> pojoTypeInfo, int[] includedFieldsMask) {
+ this(filePath, lineDelimiter, fieldDelimiter, pojoTypeInfo, pojoTypeInfo.getFieldNames(), includedFieldsMask);
+ }
+
+ public PojoCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, PojoTypeInfo<OUT> pojoTypeInfo, String[] fieldNames, int[] includedFieldsMask) {
+ super(filePath);
+ boolean[] mask = (includedFieldsMask == null)
+ ? createDefaultMask(fieldNames.length)
+ : toBooleanMask(includedFieldsMask);
+ configure(lineDelimiter, fieldDelimiter, pojoTypeInfo, fieldNames, mask);
+ }
+
+ public PojoCsvInputFormat(Path filePath, PojoTypeInfo<OUT> pojoTypeInfo, boolean[] includedFieldsMask) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, pojoTypeInfo, pojoTypeInfo.getFieldNames(), includedFieldsMask);
+ }
+
+ public PojoCsvInputFormat(Path filePath, PojoTypeInfo<OUT> pojoTypeInfo, String[] fieldNames, boolean[] includedFieldsMask) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, pojoTypeInfo, fieldNames, includedFieldsMask);
+ }
+
+ public PojoCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, PojoTypeInfo<OUT> pojoTypeInfo, boolean[] includedFieldsMask) {
+ this(filePath, lineDelimiter, fieldDelimiter, pojoTypeInfo, pojoTypeInfo.getFieldNames(), includedFieldsMask);
+ }
+
+ public PojoCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, PojoTypeInfo<OUT> pojoTypeInfo, String[] fieldNames, boolean[] includedFieldsMask) {
+ super(filePath);
+ configure(lineDelimiter, fieldDelimiter, pojoTypeInfo, fieldNames, includedFieldsMask);
+ }
+
+ private void configure(String lineDelimiter, String fieldDelimiter, PojoTypeInfo<OUT> pojoTypeInfo, String[] fieldNames, boolean[] includedFieldsMask) {
+
+ if (includedFieldsMask == null) {
+ includedFieldsMask = createDefaultMask(fieldNames.length);
+ }
+
+ for (String name : fieldNames) {
+ if (name == null) {
+ throw new NullPointerException("Field name must not be null.");
+ }
+ if (pojoTypeInfo.getFieldIndex(name) < 0) {
+ throw new IllegalArgumentException("Field \"" + name + "\" not part of POJO type " + pojoTypeInfo.getTypeClass().getCanonicalName());
+ }
+ }
+
+ setDelimiter(lineDelimiter);
+ setFieldDelimiter(fieldDelimiter);
+
+ Class<?>[] classes = new Class<?>[fieldNames.length];
+
+ for (int i = 0; i < fieldNames.length; i++) {
+ try {
+ classes[i] = pojoTypeInfo.getTypeAt(pojoTypeInfo.getFieldIndex(fieldNames[i])).getTypeClass();
+ } catch (IndexOutOfBoundsException e) {
+ throw new IllegalArgumentException("Invalid field name: " + fieldNames[i]);
+ }
+ }
+
+ this.pojoTypeClass = pojoTypeInfo.getTypeClass();
+ this.pojoTypeInfo = pojoTypeInfo;
+ setFieldsGeneric(includedFieldsMask, classes);
+ setOrderOfPOJOFields(fieldNames);
+ }
+
+ private void setOrderOfPOJOFields(String[] fieldNames) {
+ Preconditions.checkNotNull(fieldNames);
+
+ int includedCount = 0;
+ for (boolean isIncluded : fieldIncluded) {
+ if (isIncluded) {
+ includedCount++;
+ }
+ }
+
+ Preconditions.checkArgument(includedCount == fieldNames.length, includedCount
+ + " CSV fields and " + fieldNames.length + " POJO fields selected. The number of selected CSV and POJO fields must be equal.");
+
+ for (String field : fieldNames) {
+ Preconditions.checkNotNull(field, "The field name cannot be null.");
+ Preconditions.checkArgument(pojoTypeInfo.getFieldIndex(field) != -1,
+ "Field \"" + field + "\" is not a member of POJO class " + pojoTypeClass.getName());
+ }
+
+ pojoFieldNames = Arrays.copyOfRange(fieldNames, 0, fieldNames.length);
+ }
+
+ @Override
+ public void open(FileInputSplit split) throws IOException {
+ super.open(split);
+
+ pojoFields = new Field[pojoFieldNames.length];
+
+ Map<String, Field> allFields = new HashMap<String, Field>();
+
+ findAllFields(pojoTypeClass, allFields);
+
+ for (int i = 0; i < pojoFieldNames.length; i++) {
+ pojoFields[i] = allFields.get(pojoFieldNames[i]);
+
+ if (pojoFields[i] != null) {
+ pojoFields[i].setAccessible(true);
+ } else {
+ throw new RuntimeException("There is no field called \"" + pojoFieldNames[i] + "\" in " + pojoTypeClass.getName());
+ }
+ }
+ }
+
+ /**
+ * Finds all declared fields in a class and all its super classes.
+ *
+ * @param clazz Class for which all declared fields are found
+ * @param allFields Map containing all found fields so far
+ */
+ private void findAllFields(Class<?> clazz, Map<String, Field> allFields) {
+ for (Field field : clazz.getDeclaredFields()) {
+ allFields.put(field.getName(), field);
+ }
+
+ if (clazz.getSuperclass() != null) {
+ findAllFields(clazz.getSuperclass(), allFields);
+ }
+ }
+
+ @Override
+ public OUT fillRecord(OUT reuse, Object[] parsedValues) {
+ for (int i = 0; i < parsedValues.length; i++) {
+ try {
+ pojoFields[i].set(reuse, parsedValues[i]);
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException("Parsed value could not be set in POJO field \"" + pojoFieldNames[i] + "\"", e);
+ }
+ }
+ return reuse;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/main/java/org/apache/flink/api/java/io/TupleCsvInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/TupleCsvInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/io/TupleCsvInputFormat.java
new file mode 100644
index 0000000..82caddd
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/io/TupleCsvInputFormat.java
@@ -0,0 +1,90 @@
+/*
+ * 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.flink.api.java.io;
+
+import org.apache.flink.core.fs.Path;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase;
+
+public class TupleCsvInputFormat<OUT> extends CsvInputFormat<OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ private TupleSerializerBase<OUT> tupleSerializer;
+
+ public TupleCsvInputFormat(Path filePath, TupleTypeInfoBase<OUT> tupleTypeInfo) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, tupleTypeInfo);
+ }
+
+ public TupleCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, TupleTypeInfoBase<OUT> tupleTypeInfo) {
+ this(filePath, lineDelimiter, fieldDelimiter, tupleTypeInfo, createDefaultMask(tupleTypeInfo.getArity()));
+ }
+
+ public TupleCsvInputFormat(Path filePath, TupleTypeInfoBase<OUT> tupleTypeInfo, int[] includedFieldsMask) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, tupleTypeInfo, includedFieldsMask);
+ }
+
+ public TupleCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, TupleTypeInfoBase<OUT> tupleTypeInfo, int[] includedFieldsMask) {
+ super(filePath);
+ boolean[] mask = (includedFieldsMask == null)
+ ? createDefaultMask(tupleTypeInfo.getArity())
+ : toBooleanMask(includedFieldsMask);
+ configure(lineDelimiter, fieldDelimiter, tupleTypeInfo, mask);
+ }
+
+ public TupleCsvInputFormat(Path filePath, TupleTypeInfoBase<OUT> tupleTypeInfo, boolean[] includedFieldsMask) {
+ this(filePath, DEFAULT_LINE_DELIMITER, DEFAULT_FIELD_DELIMITER, tupleTypeInfo, includedFieldsMask);
+ }
+
+ public TupleCsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, TupleTypeInfoBase<OUT> tupleTypeInfo, boolean[] includedFieldsMask) {
+ super(filePath);
+ configure(lineDelimiter, fieldDelimiter, tupleTypeInfo, includedFieldsMask);
+ }
+
+ private void configure(String lineDelimiter, String fieldDelimiter,
+ TupleTypeInfoBase<OUT> tupleTypeInfo, boolean[] includedFieldsMask) {
+
+ if (tupleTypeInfo.getArity() == 0) {
+ throw new IllegalArgumentException("Tuple size must be greater than 0.");
+ }
+
+ if (includedFieldsMask == null) {
+ includedFieldsMask = createDefaultMask(tupleTypeInfo.getArity());
+ }
+
+ tupleSerializer = (TupleSerializerBase<OUT>) tupleTypeInfo.createSerializer(new ExecutionConfig());
+
+ setDelimiter(lineDelimiter);
+ setFieldDelimiter(fieldDelimiter);
+
+ Class<?>[] classes = new Class<?>[tupleTypeInfo.getArity()];
+
+ for (int i = 0; i < tupleTypeInfo.getArity(); i++) {
+ classes[i] = tupleTypeInfo.getTypeAt(i).getTypeClass();
+ }
+
+ setFieldsGeneric(includedFieldsMask, classes);
+ }
+
+ @Override
+ public OUT fillRecord(OUT reuse, Object[] parsedValues) {
+ return tupleSerializer.createOrReuseInstance(parsedValues, reuse);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java
index a2d37ce..70b9393 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java
@@ -469,19 +469,12 @@ class TupleGenerator {
// create csv input format
sb.append("\t\tCsvInputFormat<Tuple" + numFields + "<");
appendTupleTypeGenerics(sb, numFields);
- sb.append(">> inputFormat = new CsvInputFormat<Tuple" + numFields + "<");
+ sb.append(">> inputFormat = new TupleCsvInputFormat<Tuple" + numFields + "<");
appendTupleTypeGenerics(sb, numFields);
- sb.append(">>(path, types);\n");
+ sb.append(">>(path, types, this.includedMask);\n");
// configure input format
- sb.append("\t\tconfigureInputFormat(inputFormat, ");
- for (int i = 0; i < numFields; i++) {
- if (i > 0) {
- sb.append(", ");
- }
- sb.append("type" + i);
- }
- sb.append(");\n");
+ sb.append("\t\tconfigureInputFormat(inputFormat);\n");
// return
sb.append("\t\treturn new DataSource<Tuple" + numFields + "<");
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
index 46e3990..0897063 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
@@ -89,6 +89,14 @@ public class TupleSerializer<T extends Tuple> extends TupleSerializerBase<T> {
}
@Override
+ public T createOrReuseInstance(Object[] fields, T reuse) {
+ for (int i = 0; i < arity; i++) {
+ reuse.setField(fields[i], i);
+ }
+ return reuse;
+ }
+
+ @Override
public T copy(T from) {
T target = instantiateRaw();
for (int i = 0; i < arity; i++) {
http://git-wip-us.apache.org/repos/asf/flink/blob/bd61f2db/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java
index bf3c7a1..fc657a1 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java
@@ -67,6 +67,8 @@ public abstract class TupleSerializerBase<T> extends TypeSerializer<T> {
// of immutable Typles (i.e. Scala Tuples)
public abstract T createInstance(Object[] fields);
+ public abstract T createOrReuseInstance(Object[] fields, T reuse);
+
@Override
public void copy(DataInputView source, DataOutputView target) throws IOException {
for (int i = 0; i < arity; i++) {