private void uploadStream ( ) throws SnowflakeSQLException { try { threadExecutor = SnowflakeUtil . createDefaultExecutorService ( "sf-stream-upload-worker-" , 1 ) ; RemoteStoreFileEncryptionMaterial encMat = encryptionMaterial . get ( 0 ) ; if ( commandType == CommandType . UPLOAD ) { threadExecutor . submit ( getUploadFileCallable ( stageInfo , SRC_FILE_NAME_FOR_STREAM , fileMetadataMap . get ( SRC_FILE_NAME_FOR_STREAM ) , ( stageInfo . getStageType ( ) == StageInfo . StageType . LOCAL_FS ) ? null : storageFactory . createClient ( stageInfo , parallel , encMat ) , connection , command , sourceStream , true , parallel , null , encMat ) ) ; } else if ( commandType == CommandType . DOWNLOAD ) { throw new SnowflakeSQLException ( SqlState . INTERNAL_ERROR , ErrorCode . INTERNAL_ERROR . getMessageCode ( ) ) ; } threadExecutor . shutdown ( ) ; try { // wait for all threads to complete without timeout threadExecutor . awaitTermination ( Long . MAX_VALUE , TimeUnit . DAYS ) ; } catch ( InterruptedException ex ) { throw new SnowflakeSQLException ( SqlState . QUERY_CANCELED , ErrorCode . INTERRUPTED . getMessageCode ( ) ) ; } logger . debug ( "Done with uploading from a stream" ) ; } finally { if ( threadExecutor != null ) { threadExecutor . shutdownNow ( ) ; threadExecutor = null ; } } }
Helper to upload data from a stream