DRILL-6850: Allow configuring table names case sensitivity for JDBC storage plugin

closes #1542

8 files changed, 98 insertions, 18 deletions

diff --git a/contrib/pom.xml b/contrib/pom.xml
index 796e79bdf..59e12da59 100644
--- a/contrib/pom.xml
+++ b/contrib/pom.xml
@@ -31,6 +31,10 @@
   <name>contrib/Parent Pom</name>
   <packaging>pom</packaging>
 
+  <properties>
+    <skipTests>false</skipTests>
+  </properties>
+
   <dependencies>
   </dependencies>
 
diff --git a/contrib/storage-jdbc/pom.xml b/contrib/storage-jdbc/pom.xml
index 5bad6b111..9895b1aa8 100755
--- a/contrib/storage-jdbc/pom.xml
+++ b/contrib/storage-jdbc/pom.xml
@@ -34,7 +34,6 @@
     <mysql.connector.version>8.0.13</mysql.connector.version>
     <derby.database.name>drill_derby_test</derby.database.name>
     <mysql.database.name>drill_mysql_test</mysql.database.name>
-    <skipTests>false</skipTests>
   </properties>
 
   <dependencies>
@@ -112,6 +111,7 @@
             <mysql.port>${mysql.reserved.port}</mysql.port>
             <mysql.name>${mysql.database.name}</mysql.name>
           </systemPropertyVariables>
+          <skipITs>${skipTests}</skipITs>
           <includes>
             <include>**/*IT.java</include>
           </includes>
@@ -121,6 +121,15 @@
           <execution>
             <id>run-IT-Tests</id>
             <phase>integration-test</phase>
+            <goals>
+              <goal>integration-test</goal>
+            </goals>
+          </execution>
+          <execution>
+            <phase>verify</phase>
+            <goals>
+              <goal>verify</goal>
+            </goals>
           </execution>
         </executions>
       </plugin>
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcStorageConfig.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcStorageConfig.java
index 3c3ce3c44..1c607d606 100755
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcStorageConfig.java
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcStorageConfig.java
@@ -34,18 +34,21 @@ public class JdbcStorageConfig extends StoragePluginConfig {
   private final String url;
   private final String username;
   private final String password;
+  private final boolean caseInsensitiveTableNames;
 
   @JsonCreator
   public JdbcStorageConfig(
       @JsonProperty("driver") String driver,
       @JsonProperty("url") String url,
       @JsonProperty("username") String username,
-      @JsonProperty("password") String password) {
+      @JsonProperty("password") String password,
+      @JsonProperty("caseInsensitiveTableNames") boolean caseInsensitiveTableNames) {
     super();
     this.driver = driver;
     this.url = url;
     this.username = username;
     this.password = password;
+    this.caseInsensitiveTableNames = caseInsensitiveTableNames;
   }
 
   public String getDriver() {
@@ -64,6 +67,11 @@ public class JdbcStorageConfig extends StoragePluginConfig {
     return password;
   }
 
+  @JsonProperty("caseInsensitiveTableNames")
+  public boolean areTableNamesCaseInsensitive() {
+    return caseInsensitiveTableNames;
+  }
+
   @Override
   public int hashCode() {
     final int prime = 31;
@@ -72,6 +80,7 @@ public class JdbcStorageConfig extends StoragePluginConfig {
     result = prime * result + ((password == null) ? 0 : password.hashCode());
     result = prime * result + ((url == null) ? 0 : url.hashCode());
     result = prime * result + ((username == null) ? 0 : username.hashCode());
+    result = prime * result + (caseInsensitiveTableNames ? 1231 : 1237);
     return result;
   }
 
@@ -87,6 +96,9 @@ public class JdbcStorageConfig extends StoragePluginConfig {
       return false;
     }
     JdbcStorageConfig other = (JdbcStorageConfig) obj;
+    if (caseInsensitiveTableNames != other.caseInsensitiveTableNames) {
+      return false;
+    }
     if (driver == null) {
       if (other.driver != null) {
         return false;
@@ -117,6 +129,4 @@ public class JdbcStorageConfig extends StoragePluginConfig {
     }
     return true;
   }
-
-
 }
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 cd9a6c45c..ebff37173 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
@@ -292,10 +292,24 @@ public class JdbcStoragePlugin extends AbstractStoragePlugin {
       if (table != null) {
         return table;
       }
-      return inner.getTable(name.toUpperCase());
+      if (!areTableNamesCaseSensitive()) {
+        // Oracle and H2 changes unquoted identifiers to uppercase.
+        table = inner.getTable(name.toUpperCase());
+        if (table != null) {
+          return table;
+        }
+        // Postgres changes unquoted identifiers to lowercase.
+        return inner.getTable(name.toLowerCase());
+      }
 
+      // no table was found.
+      return null;
     }
 
+    @Override
+    public boolean areTableNamesCaseSensitive() {
+      return !config.areTableNamesCaseInsensitive();
+    }
   }
 
   private class JdbcCatalogSchema extends AbstractSchema {
@@ -335,8 +349,6 @@ public class JdbcStoragePlugin extends AbstractStoragePlugin {
       }
 
       defaultSchema = schemaMap.values().iterator().next();
-
-
     }
 
     void setHolder(SchemaPlus plusOfThis) {
@@ -405,14 +417,9 @@ public class JdbcStoragePlugin extends AbstractStoragePlugin {
 
       if (schema != null) {
         try {
-          Table t = schema.getTable(name);
-          if (t != null) {
-            return t;
-          }
-          return schema.getTable(name.toUpperCase());
+          return schema.getTable(name);
         } catch (RuntimeException e) {
           logger.warn("Failure while attempting to read table '{}' from JDBC source.", name, e);
-
         }
       }
 
@@ -425,6 +432,10 @@ public class JdbcStoragePlugin extends AbstractStoragePlugin {
       return defaultSchema.getTableNames();
     }
 
+    @Override
+    public boolean areTableNamesCaseSensitive() {
+      return defaultSchema.areTableNamesCaseSensitive();
+    }
   }
 
   @Override
diff --git a/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithDerbyIT.java b/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithDerbyIT.java
index a22b40aef..168b5f37a 100644
--- a/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithDerbyIT.java
+++ b/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithDerbyIT.java
@@ -41,7 +41,7 @@ public class TestJdbcPluginWithDerbyIT extends PlanTestBase {
   private static final String TABLE_NAME = String.format("%s.`%s`", StoragePluginTestUtils.DFS_PLUGIN_NAME, TABLE_PATH);
 
   @BeforeClass
-  public static void copyData() throws Exception {
+  public static void copyData() {
     dirTestWatcher.copyResourceToRoot(Paths.get(TABLE_PATH));
   }
 
@@ -127,12 +127,18 @@ public class TestJdbcPluginWithDerbyIT extends PlanTestBase {
   public void showTablesDefaultSchema() throws Exception {
     testNoResult("use derby.drill_derby_test");
     assertEquals(1, testSql("show tables like 'PERSON'"));
+
+    // check table names case insensitivity
+    assertEquals(1, testSql("show tables like 'person'"));
   }
 
   @Test
   public void describe() throws Exception {
     testNoResult("use derby.drill_derby_test");
     assertEquals(19, testSql("describe PERSON"));
+
+    // check table names case insensitivity
+    assertEquals(19, testSql("describe person"));
   }
 
   @Test
@@ -147,4 +153,16 @@ public class TestJdbcPluginWithDerbyIT extends PlanTestBase {
     String query = "select * from derby.drill_derby_test.person where person_ID = 1";
     testPlanMatchingPatterns(query, new String[]{}, new String[]{"Filter"});
   }
+
+  @Test
+  public void testCaseInsensitiveTableNames() throws Exception {
+    assertEquals(5, testSql("select * from derby.drill_derby_test.PeRsOn"));
+    assertEquals(5, testSql("select * from derby.drill_derby_test.PERSON"));
+    assertEquals(5, testSql("select * from derby.drill_derby_test.person"));
+  }
+
+  @Test
+  public void testJdbcStoragePluginSerDe() throws Exception {
+    testPhysicalPlanExecutionBasedOnQuery("select * from derby.drill_derby_test.PeRsOn");
+  }
 }
diff --git a/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithMySQLIT.java b/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithMySQLIT.java
index c8396b509..361559c93 100644
--- a/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithMySQLIT.java
+++ b/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithMySQLIT.java
@@ -26,6 +26,8 @@ import org.junit.experimental.categories.Category;
 
 import java.math.BigDecimal;
 
+import static org.junit.Assert.assertEquals;
+
 /**
  * JDBC storage plugin tests against MySQL.
  * Note: it requires libaio.so library in the system
@@ -141,5 +143,18 @@ public class TestJdbcPluginWithMySQLIT extends PlanTestBase {
     test(query);
   }
 
+  @Test
+  public void testCaseSensitiveTableNames() throws Exception {
+    test("use mysqlCaseInsensitive.`drill_mysql_test`");
+    // two table names match the filter ignoring the case
+    assertEquals(2, testSql("show tables like 'caseSensitiveTable'"));
+
+    test("use mysql.`drill_mysql_test`");
+    // single table matches the filter considering table name the case
+    assertEquals(1, testSql("show tables like 'caseSensitiveTable'"));
 
+    // checks that tables with names in different case are recognized correctly
+    assertEquals(1, testSql("describe caseSensitiveTable"));
+    assertEquals(2, testSql("describe CASESENSITIVETABLE"));
+  }
 }
diff --git a/contrib/storage-jdbc/src/test/resources/bootstrap-storage-plugins.json b/contrib/storage-jdbc/src/test/resources/bootstrap-storage-plugins.json
index 4018d9247..945ddeb53 100755
--- a/contrib/storage-jdbc/src/test/resources/bootstrap-storage-plugins.json
+++ b/contrib/storage-jdbc/src/test/resources/bootstrap-storage-plugins.json
@@ -4,15 +4,22 @@
           type    : "jdbc",
           driver  : "org.apache.derby.jdbc.ClientDriver",
           url     : "jdbc:derby://localhost:${derby.reserved.port}/memory:${derby.database.name};user=root;password=root",
+          caseInsensitiveTableNames: true,
           enabled : true
         },
         mysql : {
           type    : "jdbc",
-          enabled : true,
           driver  : "com.mysql.jdbc.Driver",
           url     : "jdbc:mysql://localhost:${mysql.reserved.port}/${mysql.database.name}?user=root&password=root&useJDBCCompliantTimezoneShift=true",
           enabled : true
-      }
+        },
+        mysqlCaseInsensitive : {
+          type    : "jdbc",
+          driver  : "com.mysql.jdbc.Driver",
+          url     : "jdbc:mysql://localhost:${mysql.reserved.port}/${mysql.database.name}?user=root&password=root&useJDBCCompliantTimezoneShift=true",
+          caseInsensitiveTableNames: true,
+          enabled : true
+        }
     }
 }
 
diff --git a/contrib/storage-jdbc/src/test/resources/mysql-test-data.sql b/contrib/storage-jdbc/src/test/resources/mysql-test-data.sql
index 6875d9959..92ad6ff57 100644
--- a/contrib/storage-jdbc/src/test/resources/mysql-test-data.sql
+++ b/contrib/storage-jdbc/src/test/resources/mysql-test-data.sql
@@ -3,12 +3,18 @@ set global time_zone = "+00:00";
 
 use drill_mysql_test;
 
-create table x (
+create table caseSensitiveTable (
   a   BLOB
 );
 
-insert into x (a) values ('this is a test');
+insert into caseSensitiveTable (a) values ('this is a test');
 
+create table CASESENSITIVETABLE (
+  a   BLOB,
+  b   BLOB
+);
+
+insert into CASESENSITIVETABLE (a, b) values ('this is a test', 'for case sensitive table names');
 
 create table person (
   person_id       INT NOT NULL AUTO_INCREMENT PRIMARY KEY,