From 42c9814069dfd519bb21c445164ef23e5fa121bc Mon Sep 17 00:00:00 2001
From: Stephan Ewen <sewen@apache.org>
Date: Mon, 13 Oct 2014 21:36:10 +0200
Subject: [PATCH] [FLINK-1148] Create a writeAsCsv(path, writemode) variant

---
 .../org/apache/flink/api/java/DataSet.java    | 37 +++++++++++++------
 1 file changed, 26 insertions(+), 11 deletions(-)

diff --git a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
index dd8c4ba812cae..bfaf6126f3cb4 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
@@ -554,9 +554,8 @@ public UnsortedGrouping<T> groupBy(String... fields) {
 	 *   {@link DataSet DataSets} on key equality and provides multiple ways to combine 
 	 *   joining elements into one DataSet.</br>
 	 * 
-	 * This method returns a {@link JoinOperatorSets} on which 
-	 *   {@link JoinOperatorSets#where()} needs to be called to define the join key of the first 
-	 *   joining (i.e., this) DataSet.
+	 * This method returns a {@link JoinOperatorSets} on which one of the {@code where} methods
+	 * can be called to define the join key of the first joining (i.e., this) DataSet.
 	 *  
 	 * @param other The other DataSet with which this DataSet is joined.
 	 * @return A JoinOperatorSets to continue the definition of the Join transformation.
@@ -576,7 +575,7 @@ public <R> JoinOperatorSets<T, R> join(DataSet<R> other) {
 	 * This method also gives the hint to the optimizer that the second DataSet to join is much
 	 *   smaller than the first one.</br>
 	 * This method returns a {@link JoinOperatorSets} on which 
-	 *   {@link JoinOperatorSets#where()} needs to be called to define the join key of the first 
+	 *   {@link JoinOperatorSets#where(String...)} needs to be called to define the join key of the first 
 	 *   joining (i.e., this) DataSet.
 	 *  
 	 * @param other The other DataSet with which this DataSet is joined.
@@ -596,9 +595,8 @@ public <R> JoinOperatorSets<T, R> joinWithTiny(DataSet<R> other) {
 	 *   joining elements into one DataSet.</br>
 	 * This method also gives the hint to the optimizer that the second DataSet to join is much
 	 *   larger than the first one.</br>
-	 * This method returns a {@link JoinOperatorSets JoinOperatorSet} on which 
-	 *   {@link JoinOperatorSets#where()} needs to be called to define the join key of the first 
-	 *   joining (i.e., this) DataSet.
+	 * This method returns a {@link JoinOperatorSets} on which one of the {@code where} methods
+	 * can be called to define the join key of the first joining (i.e., this) DataSet.
 	 *  
 	 * @param other The other DataSet with which this DataSet is joined.
 	 * @return A JoinOperatorSet to continue the definition of the Join transformation.
@@ -623,9 +621,8 @@ public <R> JoinOperatorSets<T, R> joinWithHuge(DataSet<R> other) {
 	 *   is called with an empty group for the non-existing group.</br>
 	 * The CoGroupFunction can iterate over the elements of both groups and return any number 
 	 *   of elements including none.</br>
-	 * This method returns a {@link CoGroupOperatorSets} on which 
-	 *   {@link CoGroupOperatorSets#where()} needs to be called to define the grouping key of the first 
-	 *   (i.e., this) DataSet.
+	 * This method returns a {@link JoinOperatorSets} on which one of the {@code where} methods
+	 * can be called to define the join key of the first joining (i.e., this) DataSet.
 	 * 
 	 * @param other The other DataSet of the CoGroup transformation.
 	 * @return A CoGroupOperatorSets to continue the definition of the CoGroup transformation.
@@ -1002,6 +999,24 @@ public DataSink<T> writeAsCsv(String filePath) {
 		return writeAsCsv(filePath, CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
 	}
 	
+	/**
+	 * Writes a {@link Tuple} DataSet as a CSV file to the specified location.<br/>
+	 * <b>Note: Only a Tuple DataSet can written as a CSV file.</b><br/>
+	 * For each Tuple field the result of {@link Object#toString()} is written.
+	 * Tuple fields are separated by the default field delimiter {@code "comma" (,)}.<br/>
+	 * Tuples are are separated by the newline character ({@code \n}).
+	 * 
+	 * @param filePath The path pointing to the location the CSV file is written to.
+	 * @param writeMode The behavior regarding existing files. Options are NO_OVERWRITE and OVERWRITE.
+	 * @return The DataSink that writes the DataSet.
+	 * 
+	 * @see Tuple
+	 * @see CsvOutputFormat
+	 */
+	public DataSink<T> writeAsCsv(String filePath, WriteMode writeMode) {
+		return internalWriteAsCsv(new Path(filePath),CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER, writeMode);
+	}
+	
 	/**
 	 * Writes a {@link Tuple} DataSet as a CSV file to the specified location with the specified field and line delimiters.<br/>
 	 * <b>Note: Only a Tuple DataSet can written as a CSV file.</b><br/>
@@ -1026,7 +1041,7 @@ public DataSink<T> writeAsCsv(String filePath, String rowDelimiter, String field
 	 * @param filePath The path pointing to the location the CSV file is written to.
 	 * @param rowDelimiter The row delimiter to separate Tuples.
 	 * @param fieldDelimiter The field delimiter to separate Tuple fields.
-	 * @param writeMode Control the behavior for existing files. Options are NO_OVERWRITE and OVERWRITE.
+	 * @param writeMode The behavior regarding existing files. Options are NO_OVERWRITE and OVERWRITE.
 	 * 
 	 * @see Tuple
 	 * @see CsvOutputFormat