SQOOP-1224: Enable use of Oracle Wallets with Oracle Manager

(Venkat Ranganathan via Jarek Jarcec Cecho)

src/java/org/apache/sqoop/manager/OracleManager.java

@@ -109,6 +109,13 @@ public class OracleManager
      + "ALL_CONS_COLUMNS.TABLE_NAME = ? AND "
      + "ALL_CONS_COLUMNS.OWNER = ?";
 
+  /**
+   * Query to get the current user for the DB session.   Used in case of
+   * wallet logins.
+   */
+  public static final String QUERY_GET_SESSIONUSER =
+     "SELECT USER FROM DUAL";
+
   // driver class to ensure is loaded when making db connection.
   private static final String DRIVER_CLASS = "oracle.jdbc.OracleDriver";
 
@@ -284,6 +291,7 @@ protected Connection makeConnection() throws SQLException {
     String password = options.getPassword();
     String connectStr = options.getConnectString();
 
+
     connection = CACHE.getConnection(connectStr, username);
     if (null == connection) {
       // Couldn't pull one from the cache. Get a new one.
@@ -326,6 +334,44 @@ protected Connection makeConnection() throws SQLException {
     return connection;
   }
 
+  public String getSessionUser(Connection conn) {
+    Statement stmt = null;
+    ResultSet rset = null;
+    String user = null;
+    try {
+      stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY,
+              ResultSet.CONCUR_READ_ONLY);
+      rset = stmt.executeQuery(QUERY_GET_SESSIONUSER);
+
+      if (rset.next()) {
+        user = rset.getString(1);
+      }
+      conn.commit();
+    } catch (SQLException e) {
+      try {
+        conn.rollback();
+      } catch (SQLException ex) {
+        LoggingUtils.logAll(LOG, "Failed to rollback transaction", ex);
+      }
+    } finally {
+      if (rset != null) {
+        try {
+          rset.close();
+        } catch (SQLException ex) {
+          LoggingUtils.logAll(LOG, "Failed to close resultset", ex);
+        }
+      }
+      if (stmt != null) {
+        try {
+          stmt.close();
+        } catch (SQLException ex) {
+          LoggingUtils.logAll(LOG, "Failed to close statement", ex);
+        }
+      }
+    }
+    return user;
+  }
+
   /**
    * Set session time zone.
    * @param conn      Connection object
@@ -677,11 +723,12 @@ public String[] listTables() {
     PreparedStatement pStmt = null;
     ResultSet rset = null;
     List<String> tables = new ArrayList<String>();
-    String tableOwner = this.options.getUsername();
+    String tableOwner = null;
 
 
     try {
       conn = getConnection();
+      tableOwner = getSessionUser(conn);
       pStmt = conn.prepareStatement(QUERY_LIST_TABLES,
           ResultSet.TYPE_FORWARD_ONLY,
               ResultSet.CONCUR_READ_ONLY);
@@ -862,7 +909,7 @@ public String[] getColumnNames(String tableName) {
     ResultSet rset = null;
     List<String> columns = new ArrayList<String>();
 
-    String tableOwner = this.options.getUsername();
+    String tableOwner = null;
     String shortTableName = tableName;
     int qualifierIndex = tableName.indexOf('.');
     if (qualifierIndex != -1) {
@@ -873,6 +920,10 @@ public String[] getColumnNames(String tableName) {
     try {
       conn = getConnection();
 
+      if (tableOwner == null) {
+        tableOwner = getSessionUser(conn);
+      }
+
       pStmt = conn.prepareStatement(QUERY_COLUMNS_FOR_TABLE,
                   ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
 
@@ -925,7 +976,7 @@ public String getPrimaryKey(String tableName) {
     ResultSet rset = null;
     List<String> columns = new ArrayList<String>();
 
-    String tableOwner = this.options.getUsername();
+    String tableOwner = null;
     String shortTableName = tableName;
     int qualifierIndex = tableName.indexOf('.');
     if (qualifierIndex != -1) {
@@ -936,6 +987,10 @@ public String getPrimaryKey(String tableName) {
     try {
       conn = getConnection();
 
+      if (tableOwner == null) {
+        tableOwner = getSessionUser(conn);
+      }
+
       pStmt = conn.prepareStatement(QUERY_PRIMARY_KEY_FOR_TABLE,
                   ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
       pStmt.setString(1, shortTableName);
@@ -948,7 +1003,9 @@ public String getPrimaryKey(String tableName) {
       conn.commit();
     } catch (SQLException e) {
       try {
+        if (conn != null) {
           conn.rollback();
+        }
       } catch (SQLException ex) {
         LoggingUtils.logAll(LOG, "Failed to rollback transaction", ex);
       }

src/test/com/cloudera/sqoop/manager/OracleManagerTest.java

@@ -532,4 +532,18 @@ public void testPurgeClosedConnections() throws Exception {
 
     m2.close(); // Close the manager's active connection again.
   }
+
+  public void testSessionUserName() throws Exception {
+    SqoopOptions options = new SqoopOptions(OracleUtils.CONNECT_STRING,
+      TABLE_NAME);
+    OracleUtils.setOracleAuth(options);
+
+    // Create a connection manager and get a connection
+    OracleManager m1 = new OracleManager(options);
+    Connection c1 = m1.getConnection();
+    // Make sure that the session username is the same as the Oracle
+    // sqoop user name
+    String sessionUserName = m1.getSessionUser(c1);
+    assertEquals(OracleUtils.ORACLE_USER_NAME, sessionUserName);
+  }
 }