Flink 3750 fixed #1941

fpompermaier · 2016-04-26T17:15:57Z

Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the How To Contribute guide.
In addition to going through the list, please provide a meaningful description of your changes.

General
- The pull request references the related JIRA issue
- The pull request addresses only one issue
- Each commit in the PR has a meaningful commit message
Documentation
- Documentation has been added for new functionality
- Old documentation affected by the pull request has been updated
- JavaDoc for public methods has been added
Tests & Build
- Functionality added by the pull request is covered by tests
- mvn clean verify has been executed successfully locally or a Travis build has passed

New cleaned PR aligned with the current master

fhueske · 2016-05-02T13:06:33Z

...h-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java


 	public JDBCInputFormat() {
 	}

 	@Override
 	public void configure(Configuration parameters) {
+		//called once per inputFormat (on open)
+		try {
+			establishConnection();


Use the new method in RichInputFormat (openInputFormat() and closeInputFormat())

I think I forgotto include the new connection management when I closed the previous branch..

fhueske · 2016-05-02T13:20:55Z

Thanks for the PR @fpompermaier.
I think the new format is a bit too much tailored towards certain query templates (BETWEEN predicate on integer column). Also modifying queries that users provide, is a bit risky, IMO.

To make it more general I would propose to:

Accept query templates with markers, similar to parameter makers in prepared statements: SELECT address FROM people WHERE name = ? AND birthday BETWEEN ? AND ?.
Let users explicitly provide bounds. Users should know their data best and can provide bounds which take skewed distributions into account. Parameter values can be provided as Object[], one array for each parameter. We can provide some utility methods to help users generating uniformly distributed parameter values.
Let InputSplit not provide two bound values but the index for the parameter value array. So each instance can build the query by substituting the parameters by values.

What do you think?

fhueske · 2016-05-04T10:23:09Z

...h-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/QueryParamInputSplit.java

+/**
+ * Query parameters container 
+ */
+public class QueryParamInputSplit implements InputSplit {


I don't think we need a special InputSplit.
If we store the binding values in a Object[][] member variable of the JDBCInputFormat, we only need an index to look up the respective values. So we can use GenericInputSplit which has a split number that can be used as index.

fhueske · 2016-05-04T10:41:53Z

Thanks for the update @fpompermaier. I added a few comments and suggestions.

fhueske · 2016-05-04T11:39:17Z

...flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/split/GenericSplitsGenerator.java

+ * This splits generator allows the apply user defined splits (computed outside the IF) 
+ * 
+ * */
+public class GenericSplitsGenerator implements JDBCInputSplitsGenerator {


Please rename to GenericParameterValueProvider

fhueske · 2016-05-11T11:40:10Z

...h-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java

@@ -81,25 +134,51 @@ public void configure(Configuration parameters) {
 	 * @throws IOException
 	 */
 	@Override
-	public void open(InputSplit ignored) throws IOException {
+	public void open(InputSplit inputSplit) throws IOException {


JavaDocs need to be updated for new parameter name

fhueske · 2016-05-11T11:47:36Z

...h-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java

@@ -352,6 +318,10 @@ public JDBCInputFormatBuilder setResultSetConcurrency(int resultSetConcurrency)
 			format.resultSetConcurrency = resultSetConcurrency;
 			return this;
 		}
+		public JDBCInputFormatBuilder setParametersProvider(ParameterValuesProvider parameterValuesProvider) {


Add new line above

fhueske · 2016-05-11T11:58:55Z

Thanks for the update @fpompermaier. Overall the PR looks good. I added a few comments and suggestions. Thanks, Fabian

fhueske · 2016-05-13T10:17:46Z

...-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormat.java

 	 * 
-	 * WARNING: this may fail if the JDBC driver doesn't handle null correctly and no column types specified in the SqlRow
+	 * WARNING: this may fail when no column types specified (because a best effort approach is attempted in order to


This warning should also go into the JavaDocs of the JdbcOutputFormat class.

fhueske · 2016-05-18T08:37:30Z

Hi @fpompermaier, I added one more comment. I'm not sure if you noticed the other two comments I made a fews day back to JDBCFullTest and JdbcInputFormat.nextRecord() after your last update of the PR. Thanks, Fabian

fhueske · 2016-05-18T13:32:21Z

...h-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java

- * query template (i.e. a valid {@link PreparedStatement}) and a {@link ParameterValuesProvider} 
- * which provides binding values for the query parameters.
- * 
+ * A valid RowTypeInfo must be properly configured in the builder, e.g.: </br>


Thanks for the nice JavaDocs!

fhueske · 2016-05-18T13:39:40Z

Thanks for the update @fpompermaier. Just had a few minor comments. After that this PR should be good to merge.

fhueske · 2016-05-18T17:08:11Z

Thanks for the update. Looks good. Will merge it later :-)

fhueske · 2016-05-18T19:31:44Z

Merging

…ormat. - New Input- and OutputFormat use Row instead of Tuple types to support null values. - JdbcInputFormat supports parallel input due to PreparedStatement and binding values for parameters. This closes apache#1941

fhueske reviewed May 2, 2016
View reviewed changes

fhueske reviewed May 4, 2016
View reviewed changes

fpompermaier added 5 commits May 4, 2016 17:42

FLINK-3750 new PR (the old one was messed up...)

7ddd3fc

Removed commented method

ce17dc1

Fixed connection opening and closing

122da62

Used Row to exchange data and generalized splitting strategies

2b44cc6

Implemented @fhueske's suggestions

b0dd704

fpompermaier force-pushed the FLINK-3750-fixed branch from ea58168 to b0dd704 Compare May 4, 2016 15:46

fpompermaier added 2 commits May 4, 2016 17:52

Simplified creation of RowTypeInfo using the new APIs

85c1ad3

Uncommented temporary commented lines of test...

eab26a6

fhueske reviewed May 11, 2016
View reviewed changes

Fixed @fhueske observations

315fcb1

fhueske reviewed May 13, 2016
View reviewed changes

Implemented @fhueske suggestions

89fd529

fhueske reviewed May 18, 2016
View reviewed changes

Last refinements

e6ce7f1

asfgit closed this in 09b428b May 18, 2016

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Flink 3750 fixed #1941

Flink 3750 fixed #1941

fpompermaier commented Apr 26, 2016

fhueske May 2, 2016

fpompermaier May 2, 2016

fhueske commented May 2, 2016

fhueske May 4, 2016

fhueske commented May 4, 2016

fhueske May 4, 2016

fhueske May 11, 2016

fhueske May 11, 2016

fhueske commented May 11, 2016

fhueske May 13, 2016

fhueske commented May 18, 2016

fhueske May 18, 2016

fhueske commented May 18, 2016

fhueske commented May 18, 2016

fhueske commented May 18, 2016

Flink 3750 fixed #1941

Flink 3750 fixed #1941

Conversation

fpompermaier commented Apr 26, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fhueske commented May 2, 2016

Choose a reason for hiding this comment

fhueske commented May 4, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fhueske commented May 11, 2016

Choose a reason for hiding this comment

fhueske commented May 18, 2016

Choose a reason for hiding this comment

fhueske commented May 18, 2016

fhueske commented May 18, 2016

fhueske commented May 18, 2016