code: 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;
        }
      }
    }
docs: Helper to upload data from a stream