HIVE-24387: Metastore access through JDBC handler does not use correct database accessor (Jesus Camacho Rodriguez, reviewed by Vineet Garg)

Closes #1673

Author: jcamachor

jdbc-handler/src/main/java/org/apache/hive/storage/jdbc/dao/DatabaseAccessorFactory.java

@@ -28,13 +28,14 @@ private DatabaseAccessorFactory() {
   }
 
 
-  public static DatabaseAccessor getAccessor(DatabaseType dbType) {
+  public static DatabaseAccessor getAccessor(DatabaseType dbType, String driverClass) {
 
     DatabaseAccessor accessor = null;
     switch (dbType) {
     case MYSQL:
       accessor = new MySqlDatabaseAccessor();
       break;
+
     case JETHRO_DATA:
       accessor = new JethroDatabaseAccessor();
       break;
@@ -63,6 +64,26 @@ public static DatabaseAccessor getAccessor(DatabaseType dbType) {
       accessor = new DerbyDatabaseAccessor();
       break;
 
+    case METASTORE:
+      // For metastore, we infer the accessor from the jdbc driver string.
+      // TODO: We could also make a call to get the metadata from the connection
+      //       so we could obtain the database product. However, it seems an
+      //       overkill given that the metastore supported RDBMSs is a
+      //       well-defined set.
+      if (driverClass.contains("MYSQL")) {
+        accessor = new MySqlDatabaseAccessor();
+      } else if (driverClass.contains("POSTGRESQL")) {
+        accessor = new PostgresDatabaseAccessor();
+      } else if (driverClass.contains("ORACLE")) {
+        accessor = new OracleDatabaseAccessor();
+      } else if (driverClass.contains("SQLSERVER")) {
+        accessor = new MsSqlDatabaseAccessor();
+      } else {
+        // default
+        accessor = new GenericJdbcDatabaseAccessor();
+      }
+      break;
+
     default:
       accessor = new GenericJdbcDatabaseAccessor();
       break;
@@ -75,7 +96,9 @@ public static DatabaseAccessor getAccessor(DatabaseType dbType) {
   public static DatabaseAccessor getAccessor(Configuration conf) {
     DatabaseType dbType = DatabaseType.valueOf(
         conf.get(JdbcStorageConfig.DATABASE_TYPE.getPropertyName()).toUpperCase());
-    return getAccessor(dbType);
+    String driverClass =
+        conf.get(JdbcStorageConfig.JDBC_DRIVER_CLASS.getPropertyName()).toUpperCase();
+    return getAccessor(dbType, driverClass);
   }
 
 }