flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [3/3] flink git commit: [FLINK-3472] [jdbc] Give a better exception if jdbc column has a null value
Date Sun, 13 Mar 2016 16:29:55 GMT
[FLINK-3472] [jdbc] Give a better exception if jdbc column has a null value

This closes #1772


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3d8c7c1e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3d8c7c1e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3d8c7c1e

Branch: refs/heads/release-1.0
Commit: 3d8c7c1ea0791a1f3243d8f0094cb2947c028674
Parents: 43e95f2
Author: zentol <chesnay@apache.org>
Authored: Wed Mar 9 11:11:31 2016 +0100
Committer: Stephan Ewen <sewen@apache.org>
Committed: Sun Mar 13 17:29:03 2016 +0100

----------------------------------------------------------------------
 .../flink/api/java/io/jdbc/JDBCInputFormat.java | 168 ++++++++++---------
 1 file changed, 86 insertions(+), 82 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/3d8c7c1e/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
b/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
index 132edc4..b764350 100644
--- a/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
+++ b/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
@@ -191,89 +191,93 @@ public class JDBCInputFormat<OUT extends Tuple> extends RichInputFormat<OUT,
Inp
 	 *
 	 * @param reuse Target Record.
 	 */
-	private void addValue(OUT reuse) throws SQLException {
+	private void addValue(OUT reuse) throws IOException, SQLException {
 		for (int pos = 0; pos < columnTypes.length; pos++) {
-			switch (columnTypes[pos]) {
-				case java.sql.Types.NULL:
-					reuse.setField(NullValue.getInstance(), pos);
-					break;
-				case java.sql.Types.BOOLEAN:
-					reuse.setField(resultSet.getBoolean(pos + 1), pos);
-					break;
-				case java.sql.Types.BIT:
-					reuse.setField(resultSet.getBoolean(pos + 1), pos);
-					break;
-				case java.sql.Types.CHAR:
-					reuse.setField(resultSet.getString(pos + 1), pos);
-					break;
-				case java.sql.Types.NCHAR:
-					reuse.setField(resultSet.getString(pos + 1), pos);
-					break;
-				case java.sql.Types.VARCHAR:
-					reuse.setField(resultSet.getString(pos + 1), pos);
-					break;
-				case java.sql.Types.LONGVARCHAR:
-					reuse.setField(resultSet.getString(pos + 1), pos);
-					break;
-				case java.sql.Types.LONGNVARCHAR:
-					reuse.setField(resultSet.getString(pos + 1), pos);
-					break;
-				case java.sql.Types.TINYINT:
-					reuse.setField(resultSet.getShort(pos + 1), pos);
-					break;
-				case java.sql.Types.SMALLINT:
-					reuse.setField(resultSet.getShort(pos + 1), pos);
-					break;
-				case java.sql.Types.BIGINT:
-					reuse.setField(resultSet.getLong(pos + 1), pos);
-					break;
-				case java.sql.Types.INTEGER:
-					reuse.setField(resultSet.getInt(pos + 1), pos);
-					break;
-				case java.sql.Types.FLOAT:
-					reuse.setField(resultSet.getDouble(pos + 1), pos);
-					break;
-				case java.sql.Types.REAL:
-					reuse.setField(resultSet.getFloat(pos + 1), pos);
-					break;
-				case java.sql.Types.DOUBLE:
-					reuse.setField(resultSet.getDouble(pos + 1), pos);
-					break;
-				case java.sql.Types.DECIMAL:
-					reuse.setField(resultSet.getBigDecimal(pos + 1).doubleValue(), pos);
-					break;
-				case java.sql.Types.NUMERIC:
-					reuse.setField(resultSet.getBigDecimal(pos + 1).doubleValue(), pos);
-					break;
-				case java.sql.Types.DATE:
-					reuse.setField(resultSet.getDate(pos + 1).toString(), pos);
-					break;
-				case java.sql.Types.TIME:
-					reuse.setField(resultSet.getTime(pos + 1).getTime(), pos);
-					break;
-				case java.sql.Types.TIMESTAMP:
-					reuse.setField(resultSet.getTimestamp(pos + 1).toString(), pos);
-					break;
-				case java.sql.Types.SQLXML:
-					reuse.setField(resultSet.getSQLXML(pos + 1).toString(), pos);
-					break;
-				default:
-					throw new SQLException("Unsupported sql-type [" + columnTypes[pos] + "] on column ["
+ pos + "]");
-
-				// case java.sql.Types.BINARY:
-				// case java.sql.Types.VARBINARY:
-				// case java.sql.Types.LONGVARBINARY:
-				// case java.sql.Types.ARRAY:
-				// case java.sql.Types.JAVA_OBJECT:
-				// case java.sql.Types.BLOB:
-				// case java.sql.Types.CLOB:
-				// case java.sql.Types.NCLOB:
-				// case java.sql.Types.DATALINK:
-				// case java.sql.Types.DISTINCT:
-				// case java.sql.Types.OTHER:
-				// case java.sql.Types.REF:
-				// case java.sql.Types.ROWID:
-				// case java.sql.Types.STRUCT:
+			try {
+				switch (columnTypes[pos]) {
+					case java.sql.Types.NULL:
+						reuse.setField(NullValue.getInstance(), pos);
+						break;
+					case java.sql.Types.BOOLEAN:
+						reuse.setField(resultSet.getBoolean(pos + 1), pos);
+						break;
+					case java.sql.Types.BIT:
+						reuse.setField(resultSet.getBoolean(pos + 1), pos);
+						break;
+					case java.sql.Types.CHAR:
+						reuse.setField(resultSet.getString(pos + 1), pos);
+						break;
+					case java.sql.Types.NCHAR:
+						reuse.setField(resultSet.getString(pos + 1), pos);
+						break;
+					case java.sql.Types.VARCHAR:
+						reuse.setField(resultSet.getString(pos + 1), pos);
+						break;
+					case java.sql.Types.LONGVARCHAR:
+						reuse.setField(resultSet.getString(pos + 1), pos);
+						break;
+					case java.sql.Types.LONGNVARCHAR:
+						reuse.setField(resultSet.getString(pos + 1), pos);
+						break;
+					case java.sql.Types.TINYINT:
+						reuse.setField(resultSet.getShort(pos + 1), pos);
+						break;
+					case java.sql.Types.SMALLINT:
+						reuse.setField(resultSet.getShort(pos + 1), pos);
+						break;
+					case java.sql.Types.BIGINT:
+						reuse.setField(resultSet.getLong(pos + 1), pos);
+						break;
+					case java.sql.Types.INTEGER:
+						reuse.setField(resultSet.getInt(pos + 1), pos);
+						break;
+					case java.sql.Types.FLOAT:
+						reuse.setField(resultSet.getDouble(pos + 1), pos);
+						break;
+					case java.sql.Types.REAL:
+						reuse.setField(resultSet.getFloat(pos + 1), pos);
+						break;
+					case java.sql.Types.DOUBLE:
+						reuse.setField(resultSet.getDouble(pos + 1), pos);
+						break;
+					case java.sql.Types.DECIMAL:
+						reuse.setField(resultSet.getBigDecimal(pos + 1).doubleValue(), pos);
+						break;
+					case java.sql.Types.NUMERIC:
+						reuse.setField(resultSet.getBigDecimal(pos + 1).doubleValue(), pos);
+						break;
+					case java.sql.Types.DATE:
+						reuse.setField(resultSet.getDate(pos + 1).toString(), pos);
+						break;
+					case java.sql.Types.TIME:
+						reuse.setField(resultSet.getTime(pos + 1).getTime(), pos);
+						break;
+					case java.sql.Types.TIMESTAMP:
+						reuse.setField(resultSet.getTimestamp(pos + 1).toString(), pos);
+						break;
+					case java.sql.Types.SQLXML:
+						reuse.setField(resultSet.getSQLXML(pos + 1).toString(), pos);
+						break;
+					default:
+						throw new SQLException("Unsupported sql-type [" + columnTypes[pos] + "] on column ["
+ pos + "]");
+
+						// case java.sql.Types.BINARY:
+						// case java.sql.Types.VARBINARY:
+						// case java.sql.Types.LONGVARBINARY:
+						// case java.sql.Types.ARRAY:
+						// case java.sql.Types.JAVA_OBJECT:
+						// case java.sql.Types.BLOB:
+						// case java.sql.Types.CLOB:
+						// case java.sql.Types.NCLOB:
+						// case java.sql.Types.DATALINK:
+						// case java.sql.Types.DISTINCT:
+						// case java.sql.Types.OTHER:
+						// case java.sql.Types.REF:
+						// case java.sql.Types.ROWID:
+						// case java.sql.Types.STRUCT:
+				}
+			} catch (NullPointerException npe) {
+				throw new IOException("Encountered null value for column " + pos + ". Decimal, Numeric,
Date, Time, Timestamp and SQLXML columns may not contain NULL values.");
 			}
 		}
 	}


Mime
View raw message