Wrap and re-throw Await.result exceptions in order to capture full stacktrace

Exceptions thrown from Scala's `Await.result` don't include the waiting thread's stacktrace, making it hard to figure out where errors occur. Similar to the fix implemented in Spark in apache/spark#12433, this patch modifies our `Await.result` usages to wrap and rethrow exceptions to capture the calling thread's stack.

Author: Josh Rosen <joshrosen@databricks.com>

Closes #299 from JoshRosen/better-error-reporting.

Author: JoshRosen

src/it/scala/com/databricks/spark/redshift/IAMIntegrationSuite.scala

@@ -68,7 +68,7 @@ class IAMIntegrationSuite extends IntegrationSuiteBase {
           .mode(SaveMode.ErrorIfExists)
           .save()
       }
-      assert(err.getMessage.contains("is not authorized to assume IAM Role"))
+      assert(err.getCause.getMessage.contains("is not authorized to assume IAM Role"))
     } finally {
       conn.prepareStatement(s"drop table if exists $tableName").executeUpdate()
       conn.commit()

src/main/scala/com/databricks/spark/redshift/RedshiftJDBCWrapper.scala

@@ -26,6 +26,7 @@ import scala.collection.JavaConverters._
 import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.concurrent.duration.Duration
 import scala.util.Try
+import scala.util.control.NonFatal
 
 import org.apache.spark.SPARK_VERSION
 import org.apache.spark.sql.execution.datasources.jdbc.DriverRegistry
@@ -124,7 +125,16 @@ private[redshift] class JDBCWrapper {
       op: PreparedStatement => T): T = {
     try {
       val future = Future[T](op(statement))(ec)
-      Await.result(future, Duration.Inf)
+      try {
+        Await.result(future, Duration.Inf)
+      } catch {
+        case e: SQLException =>
+          // Wrap and re-throw so that this thread's stacktrace appears to the user.
+          throw new SQLException("Exception thrown in awaitResult: ", e)
+        case NonFatal(t) =>
+          // Wrap and re-throw so that this thread's stacktrace appears to the user.
+          throw new Exception("Exception thrown in awaitResult: ", t)
+      }
     } catch {
       case e: InterruptedException =>
         try {