private void uploadFiles ( Set < String > fileList , int parallel ) throws SnowflakeSQLException { try { threadExecutor = SnowflakeUtil . createDefaultExecutorService ( "sf-file-upload-worker-" , parallel ) ; for ( String srcFile : fileList ) { FileMetadata fileMetadata = fileMetadataMap . get ( srcFile ) ; // Check if the result status is already set so that we don't need to // upload it if ( fileMetadata . resultStatus != ResultStatus . UNKNOWN ) { logger . debug ( "Skipping {}, status: {}, details: {}" , srcFile , fileMetadata . resultStatus , fileMetadata . errorDetails ) ; continue ; } /** * For small files, we upload files in parallel, so we don't * want the remote store uploader to upload parts in parallel for each file. * For large files, we upload them in serial, and we want remote store uploader * to upload parts in parallel for each file. This is the reason * for the parallel value. */ File srcFileObj = new File ( srcFile ) ; threadExecutor . submit ( getUploadFileCallable ( stageInfo , srcFile , fileMetadata , ( stageInfo . getStageType ( ) == StageInfo . StageType . LOCAL_FS ) ? null : storageFactory . createClient ( stageInfo , parallel , encryptionMaterial . get ( 0 ) ) , connection , command , null , false , ( parallel > 1 ? 1 : this . parallel ) , srcFileObj , encryptionMaterial . get ( 0 ) ) ) ; logger . debug ( "submitted copy job for: {}" , srcFile ) ; } // shut down the thread executor 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" ) ; } finally { // shut down the thread pool in any case if ( threadExecutor != null ) { threadExecutor . shutdownNow ( ) ; threadExecutor = null ; } } }
This method create a thread pool based on requested number of threads and upload the files using the thread pool .