DRILL-4031: Log warning and ignore columns returned from JDBC source that are unsupported.

This closes #240

2 files changed, 47 insertions, 10 deletions

diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java
index 436f06576..e749c796c 100755
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java
@@ -17,6 +17,7 @@
  */
 package org.apache.drill.exec.store.jdbc;
 
+import java.lang.reflect.Field;
 import java.math.BigDecimal;
 import java.sql.Connection;
 import java.sql.Date;
@@ -122,6 +123,23 @@ class JdbcRecordReader extends AbstractRecordReader {
         .build();
   }
 
+  private static String nameFromType(int javaSqlType) {
+    try {
+      for (Field f : java.sql.Types.class.getFields()) {
+        if (java.lang.reflect.Modifier.isStatic(f.getModifiers()) &&
+            f.getType() == int.class &&
+            f.getInt(null) == javaSqlType) {
+          return f.getName();
+
+        }
+      }
+    } catch (IllegalArgumentException | IllegalAccessException e) {
+    }
+
+    return Integer.toString(javaSqlType);
+
+  }
+
   private Copier<?> getCopier(int jdbcType, int offset, ResultSet result, ValueVector v) {
 
     if (v instanceof NullableBigIntVector) {
@@ -170,13 +188,20 @@ class JdbcRecordReader extends AbstractRecordReader {
         final int scale = meta.getScale(i);
         MinorType minorType = JDBC_TYPE_MAPPINGS.get(jdbcType);
         if (minorType == null) {
-          throw UserException.dataReadError()
-              .message("The JDBC storage plugin failed while trying to execute a query. "
-                  + "The JDBC data type %d is not currently supported.", jdbcType)
 
+          logger.warn("Ignoring column that is unsupported.", UserException
+              .unsupportedError()
+              .message(
+                  "A column you queried has a data type that is not currently supported by the JDBC storage plugin. "
+                      + "The column's name was %s and its JDBC data type was %s. ",
+                  name,
+                  nameFromType(jdbcType))
               .addContext("sql", sql)
+              .addContext("column Name", name)
               .addContext("plugin", storagePluginName)
-              .build(logger);
+              .build(logger));
+
+          continue;
         }
 
         final MajorType type = Types.optional(minorType);
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcStoragePlugin.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcStoragePlugin.java
index 24d1f9d46..c22979f67 100755
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcStoragePlugin.java
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcStoragePlugin.java
@@ -61,6 +61,7 @@ import org.apache.drill.exec.store.SchemaConfig;
 import org.apache.drill.exec.store.jdbc.DrillJdbcRuleBase.DrillJdbcFilterRule;
 import org.apache.drill.exec.store.jdbc.DrillJdbcRuleBase.DrillJdbcProjectRule;
 
+import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
@@ -281,6 +282,10 @@ public class JdbcStoragePlugin extends AbstractStoragePlugin {
       return inner.getTableNames();
     }
 
+    public String toString() {
+      return Joiner.on(".").join(getSchemaPath());
+    }
+
     @Override
     public Table getTable(String name) {
       Table table = inner.getTable(name);
@@ -395,15 +400,22 @@ public class JdbcStoragePlugin extends AbstractStoragePlugin {
     @Override
     public Table getTable(String name) {
       Schema schema = getDefaultSchema();
+
       if (schema != null) {
-        Table t = schema.getTable(name);
-        if (t != null) {
-          return t;
+        try {
+          Table t = schema.getTable(name);
+          if (t != null) {
+            return t;
+          }
+          return schema.getTable(name.toUpperCase());
+        } catch (RuntimeException e) {
+          logger.warn("Failure while attempting to read table '{}' from JDBC source.", name, e);
+
         }
-        return schema.getTable(name.toUpperCase());
-      } else {
-        return null;
       }
+
+      // no table was found.
+      return null;
     }
 
     @Override