Merge pull request snowflakedb#103 from binglihub/beta

fix bug: creating temporary stage in data sink
smoy · Feb 13, 2019 · 272889f · 272889f
2 parents 237e843 + e222c13
commit 272889f
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/src/main/scala/net/snowflake/spark/snowflake_beta/SnowflakeSink.scala b/src/main/scala/net/snowflake/spark/snowflake_beta/SnowflakeSink.scala
@@ -55,7 +55,11 @@ class SnowflakeSink(
 
   private implicit val conn = DefaultJDBCWrapper.getConnector(param)
 
-  private val stageName: String = param.streamingStage.get
+  private val stageName: String = {
+    val name = param.streamingStage.get
+    conn.createStage(name, overwrite = false, temporary = false)
+    name
+  }
 
   private implicit val storage: CloudStorage =
     CloudStorageOperations.createStorageClient(param, conn, false, Some(stageName))._1

diff --git a/src/main/scala/net/snowflake/spark/snowflake_beta/io/CloudStorageOperations.scala b/src/main/scala/net/snowflake/spark/snowflake_beta/io/CloudStorageOperations.scala
@@ -66,7 +66,7 @@ object CloudStorageOperations {
   private final val AZ_KEY = "EncryptedKey"
   private final val AZ_MATDESC = "matdesc"
 
-  private val log = LoggerFactory.getLogger(getClass)
+  val log = LoggerFactory.getLogger(getClass)
 
   private[io] final def getDecryptedStream(
                                             stream: InputStream,
@@ -427,6 +427,11 @@ sealed trait CloudStorage {
           outputStream.write('\n')
         }
         outputStream.close()
+
+        CloudStorageOperations.log.info(
+          s"upload file $directory/$fileName to stage"
+        )
+
         new SingleElementIterator(s"$directory/$fileName")
       }
     }