private void filterExistingFiles() throws SnowflakeSQLException
  {
    /*
     * Build a reverse map from destination file name to source file path
     * The map will be used for looking up the source file for destination
     * files that already exist in destination location and mark them to be
     * skipped for uploading/downloading
     */
    Map<String, String> destFileNameToSrcFileMap =
        new HashMap<String, String>(fileMetadataMap.size());

    logger.debug("Build reverse map from destination file name to source file");

    for (Map.Entry<String, FileMetadata> entry : fileMetadataMap.entrySet())
    {
      if (entry.getValue().destFileName != null)
      {
        String prevSrcFile =
            destFileNameToSrcFileMap.put(entry.getValue().destFileName,
                                         entry.getKey());

        if (prevSrcFile != null)
        {
          FileMetadata prevFileMetadata = fileMetadataMap.get(prevSrcFile);

          prevFileMetadata.resultStatus = ResultStatus.COLLISION;
          prevFileMetadata.errorDetails = prevSrcFile + " has same name as " +
                                          entry.getKey();
        }
      }
      else
      {
        logger.debug("No dest file name found for: {}", entry.getKey());
        logger.debug("Status: {}", entry.getValue().resultStatus);
      }
    }

    // no files to be processed
    if (destFileNameToSrcFileMap.size() == 0)
    {
      return;
    }

    // determine greatest common prefix for all stage file names so that
    // we can call remote store API to list the objects and get their digest to compare
    // with local files
    String[] stageFileNames;

    if (commandType == CommandType.UPLOAD)
    {
      stageFileNames = destFileNameToSrcFileMap.keySet().toArray(new String[0]);
    }
    else
    {
      stageFileNames =
          destFileNameToSrcFileMap.values().toArray(new String[0]);
    }

    // find greatest common prefix for all stage file names
    Arrays.sort(stageFileNames);

    String greatestCommonPrefix =
        SnowflakeUtil.greatestCommonPrefix(stageFileNames[0],
                                           stageFileNames[stageFileNames.length - 1]);

    logger.debug("Greatest common prefix: {}", greatestCommonPrefix);

    // use the greatest common prefix to list objects under stage location
    if (stageInfo.getStageType() == StageInfo.StageType.S3 ||
        stageInfo.getStageType() == StageInfo.StageType.AZURE)
    {
      logger.debug("check existing files on remote storage for the common prefix");

      remoteLocation storeLocation = extractLocationAndPath(stageInfo.getLocation());

      StorageObjectSummaryCollection objectSummaries = null;

      int retryCount = 0;

      do
      {
        try
        {
          objectSummaries = storageClient.listObjects(storeLocation.location,
                                                      SnowflakeUtil.concatFilePathNames(
                                                          storeLocation.path,
                                                          greatestCommonPrefix, "/"));

          // exit retry loop
          break;
        }
        catch (Exception ex)
        {
          logger.debug("Listing objects for filtering encountered exception: {}",
                       ex.getMessage());

          storageClient.handleStorageException(ex, ++retryCount, "listObjects", connection, command);
        }
      }
      while (retryCount <= storageClient.getMaxRetries());

      for (StorageObjectSummary obj : objectSummaries)
      {
        logger.debug(
            "Existing object: key={} size={} md5={}",
            obj.getKey(), obj.getSize(), obj.getMD5());

        int idxOfLastFileSep = obj.getKey().lastIndexOf("/");
        String objFileName = obj.getKey().substring(idxOfLastFileSep + 1);

        // get the path to the local file so that we can calculate digest
        String mappedSrcFile = destFileNameToSrcFileMap.get(objFileName);

        // skip objects that don't have a corresponding file to be uploaded
        if (mappedSrcFile == null)
        {
          continue;
        }

        logger.debug("Next compare digest for {} against {} on the remote store", mappedSrcFile, objFileName);

        String localFile = null;
        final boolean remoteEncrypted;

        try
        {
          localFile = (commandType == CommandType.UPLOAD) ?
                      mappedSrcFile : (localLocation + objFileName);

          if (commandType == CommandType.DOWNLOAD &&
              !(new File(localFile)).exists())
          {
            logger.debug("File does not exist locally, will download {}",
                         mappedSrcFile);
            continue;
          }

          // Check file size first, if their difference is bigger than the block
          // size, we don't need to compare digests
          if (!fileMetadataMap.get(mappedSrcFile).requireCompress &&
              Math.abs(obj.getSize() - (new File(localFile)).length()) > 16)
          {
            logger.debug("Size diff between remote and local, will {} {}",
                         commandType.name().toLowerCase(), mappedSrcFile);
            continue;
          }

          // Get object metadata from remote storage
          //
          StorageObjectMetadata meta;

          try
          {
            meta = storageClient.getObjectMetadata(obj.getLocation(),
                                                   obj.getKey());
          }
          catch (StorageProviderException spEx)
          {
            // SNOW-14521: when file is not found, ok to upload
            if (spEx.isServiceException404())
            {
              // log it
              logger.debug("File returned from listing but found missing {} when getting its" +
                           " metadata. Location={}, key={}",
                           obj.getLocation(), obj.getKey());

              // the file is not found, ok to upload
              continue;
            }


            // for any other exception, log an error
            logger.error("Fetching object metadata encountered exception: {}",
                         spEx.getMessage());

            throw spEx;
          }

          String objDigest = storageClient.getDigestMetadata(meta);

          remoteEncrypted = MatDesc.parse(
              meta.getUserMetadata().get(storageClient.getMatdescKey())) != null;

          // calculate the digest hash of the local file
          InputStream fileStream = null;
          String hashText = null;

          // Streams (potentially with temp files) to clean up
          final List<FileBackedOutputStream> fileBackedOutputStreams
              = new ArrayList<>();
          try
          {
            fileStream = new FileInputStream(localFile);
            if (fileMetadataMap.get(mappedSrcFile).requireCompress)
            {
              logger.debug("Compressing stream for digest check");

              InputStreamWithMetadata res = compressStreamWithGZIP(fileStream);

              fileStream =
                  res.fileBackedOutputStream.asByteSource().openStream();
              fileBackedOutputStreams.add(res.fileBackedOutputStream);
            }

            // If the remote file has our digest, compute the SHA-256
            // for the local file
            // If the remote file does not have our digest but is unencrypted,
            // we compare the MD5 of the unencrypted local file to the ETag
            // of the S3 file.
            // Otherwise (remote file is encrypted, but has no sfc-digest),
            // no comparison is performed
            if (objDigest != null)
            {
              InputStreamWithMetadata res = computeDigest(fileStream, false);
              hashText = res.digest;
              fileBackedOutputStreams.add(res.fileBackedOutputStream);

            }
            else if (!remoteEncrypted)
            {
              hashText = DigestUtils.md5Hex(fileStream);
            }
          }
          finally
          {
            if (fileStream != null)
            {
              fileStream.close();
            }

            for (FileBackedOutputStream stream : fileBackedOutputStreams)
            {
              if (stream != null)
              {
                try
                {
                  stream.reset();
                }
                catch (IOException ex)
                {
                  logger.debug("failed to clean up temp file: {}", ex);
                }
              }
            }
          }

          // continue so that we will upload the file
          if (hashText == null || // remote is encrypted & has no digest
              (objDigest != null && !hashText.equals(objDigest)) || // digest mismatch
              (objDigest == null && !hashText.equals(obj.getMD5()))) // ETag/MD5 mismatch
          {
            logger.debug(
                "digest diff between remote store and local, will {} {}, " +
                "local digest: {}, remote store md5: {}",
                commandType.name().toLowerCase(),
                mappedSrcFile, hashText, obj.getMD5());
            continue;
          }
        }
        catch (IOException | NoSuchAlgorithmException ex)
        {
          throw new SnowflakeSQLException(ex, SqlState.INTERNAL_ERROR,
                                          ErrorCode.INTERNAL_ERROR.getMessageCode(),
                                          "Error reading: " + localFile);
        }

        logger.debug("digest same between remote store and local, will not upload {} {}",
                     commandType.name().toLowerCase(), mappedSrcFile);

        skipFile(mappedSrcFile, objFileName);
      }
    }
    else if (stageInfo.getStageType() == StageInfo.StageType.LOCAL_FS)
    {
      for (String stageFileName : stageFileNames)
      {
        String stageFilePath =
            SnowflakeUtil.concatFilePathNames(stageInfo.getLocation(),
                                              stageFileName,
                                              localFSFileSep);

        File stageFile = new File(stageFilePath);

        // if stage file doesn't exist, no need to skip whether for
        // upload/download
        if (!stageFile.exists())
        {
          continue;
        }

        String mappedSrcFile = (commandType == CommandType.UPLOAD) ?
                               destFileNameToSrcFileMap.get(stageFileName) :
                               stageFileName;

        String localFile = (commandType == CommandType.UPLOAD) ?
                           mappedSrcFile :
                           (localLocation +
                            fileMetadataMap.get(mappedSrcFile).destFileName);

        // Check file size first, if they are different, we don't need
        // to check digest
        if (!fileMetadataMap.get(mappedSrcFile).requireCompress &&
            stageFile.length() != (new File(localFile)).length())
        {
          logger.debug("Size diff between stage and local, will {} {}",
                       commandType.name().toLowerCase(), mappedSrcFile);
          continue;
        }

        // stage file eixst and either we will be compressing or
        // the dest file has same size as the source file size we will
        // compare digest values below
        String localFileHashText = null;
        String stageFileHashText = null;

        List<FileBackedOutputStream> fileBackedOutputStreams = new ArrayList();
        InputStream localFileStream = null;
        try
        {
          // calculate the digest hash of the local file
          localFileStream = new FileInputStream(localFile);

          if (fileMetadataMap.get(mappedSrcFile).requireCompress)
          {
            logger.debug("Compressing stream for digest check");

            InputStreamWithMetadata res =
                compressStreamWithGZIP(localFileStream);
            fileBackedOutputStreams.add(res.fileBackedOutputStream);

            localFileStream =
                res.fileBackedOutputStream.asByteSource().openStream();
          }

          InputStreamWithMetadata res = computeDigest(localFileStream, false);
          localFileHashText = res.digest;
          fileBackedOutputStreams.add(res.fileBackedOutputStream);
        }
        catch (IOException | NoSuchAlgorithmException ex)
        {
          throw new SnowflakeSQLException(ex, SqlState.INTERNAL_ERROR,
                                          ErrorCode.INTERNAL_ERROR.getMessageCode(),
                                          "Error reading local file: " + localFile);
        }
        finally
        {
          for (FileBackedOutputStream stream : fileBackedOutputStreams)
          {
            if (stream != null)
            {
              try
              {
                stream.reset();
              }
              catch (IOException ex)
              {
                logger.debug("failed to clean up temp file: {}", ex);
              }
            }
          }
          IOUtils.closeQuietly(localFileStream);
        }

        FileBackedOutputStream fileBackedOutputStream = null;
        InputStream stageFileStream = null;
        try
        {
          // calculate digst for stage file
          stageFileStream = new FileInputStream(stageFilePath);

          InputStreamWithMetadata res = computeDigest(stageFileStream, false);
          stageFileHashText = res.digest;
          fileBackedOutputStream = res.fileBackedOutputStream;

        }
        catch (IOException | NoSuchAlgorithmException ex)
        {
          throw new SnowflakeSQLException(ex, SqlState.INTERNAL_ERROR,
                                          ErrorCode.INTERNAL_ERROR.getMessageCode(),
                                          "Error reading stage file: " + stageFilePath);
        }
        finally
        {
          try
          {
            if (fileBackedOutputStream != null)
            {
              fileBackedOutputStream.reset();
            }
          }
          catch (IOException ex)
          {
            logger.debug("failed to clean up temp file: {}", ex);
          }
          IOUtils.closeQuietly(stageFileStream);
        }

        // continue if digest is different so that we will process the file
        if (!stageFileHashText.equals(localFileHashText))
        {
          logger.debug("digest diff between local and stage, will {} {}",
                       commandType.name().toLowerCase(), mappedSrcFile);
          continue;
        }
        else
        {
          logger.debug("digest matches between local and stage, will skip {}",
                       mappedSrcFile);

          // skip the file given that the check sum is the same b/w source
          // and destination
          skipFile(mappedSrcFile, stageFileName);
        }
      }
    }
  }
private void filterExistingFiles ( ) throws SnowflakeSQLException { Map < String , String > destFileNameToSrcFileMap = new HashMap < String , String > ( fileMetadataMap . size ( ) ) ; logger . debug ( "Build reverse map from destination file name to source file" ) ; for ( Map . Entry < String , FileMetadata > entry : fileMetadataMap . entrySet ( ) ) { if ( entry . getValue ( ) . destFileName != null ) { String prevSrcFile = destFileNameToSrcFileMap . put ( entry . getValue ( ) . destFileName , entry . getKey ( ) ) ; if ( prevSrcFile != null ) { FileMetadata prevFileMetadata = fileMetadataMap . get ( prevSrcFile ) ; prevFileMetadata . resultStatus = ResultStatus . COLLISION ; prevFileMetadata . errorDetails = prevSrcFile + " has same name as " + entry . getKey ( ) ; } } else { logger . debug ( "No dest file name found for: {}" , entry . getKey ( ) ) ; logger . debug ( "Status: {}" , entry . getValue ( ) . resultStatus ) ; } } if ( destFileNameToSrcFileMap . size ( ) == 0 ) { return ; } String [ ] stageFileNames ; if ( commandType == CommandType . UPLOAD ) { stageFileNames = destFileNameToSrcFileMap . keySet ( ) . toArray ( new String [ 0 ] ) ; } else { stageFileNames = destFileNameToSrcFileMap . values ( ) . toArray ( new String [ 0 ] ) ; } Arrays . sort ( stageFileNames ) ; String greatestCommonPrefix = SnowflakeUtil . greatestCommonPrefix ( stageFileNames [ 0 ] , stageFileNames [ stageFileNames . length - 1 ] ) ; logger . debug ( "Greatest common prefix: {}" , greatestCommonPrefix ) ; if ( stageInfo . getStageType ( ) == StageInfo . StageType . S3 || stageInfo . getStageType ( ) == StageInfo . StageType . AZURE ) { logger . debug ( "check existing files on remote storage for the common prefix" ) ; remoteLocation storeLocation = extractLocationAndPath ( stageInfo . getLocation ( ) ) ; StorageObjectSummaryCollection objectSummaries = null ; int retryCount = 0 ; do { try { objectSummaries = storageClient . listObjects ( storeLocation . location , SnowflakeUtil . concatFilePathNames ( storeLocation . path , greatestCommonPrefix , "/" ) ) ; break ; } catch ( Exception ex ) { logger . debug ( "Listing objects for filtering encountered exception: {}" , ex . getMessage ( ) ) ; storageClient . handleStorageException ( ex , ++ retryCount , "listObjects" , connection , command ) ; } } while ( retryCount <= storageClient . getMaxRetries ( ) ) ; for ( StorageObjectSummary obj : objectSummaries ) { logger . debug ( "Existing object: key={} size={} md5={}" , obj . getKey ( ) , obj . getSize ( ) , obj . getMD5 ( ) ) ; int idxOfLastFileSep = obj . getKey ( ) . lastIndexOf ( "/" ) ; String objFileName = obj . getKey ( ) . substring ( idxOfLastFileSep + 1 ) ; String mappedSrcFile = destFileNameToSrcFileMap . get ( objFileName ) ; if ( mappedSrcFile == null ) { continue ; } logger . debug ( "Next compare digest for {} against {} on the remote store" , mappedSrcFile , objFileName ) ; String localFile = null ; final boolean remoteEncrypted ; try { localFile = ( commandType == CommandType . UPLOAD ) ? mappedSrcFile : ( localLocation + objFileName ) ; if ( commandType == CommandType . DOWNLOAD && ! ( new File ( localFile ) ) . exists ( ) ) { logger . debug ( "File does not exist locally, will download {}" , mappedSrcFile ) ; continue ; } if ( ! fileMetadataMap . get ( mappedSrcFile ) . requireCompress && Math . abs ( obj . getSize ( ) - ( new File ( localFile ) ) . length ( ) ) > 16 ) { logger . debug ( "Size diff between remote and local, will {} {}" , commandType . name ( ) . toLowerCase ( ) , mappedSrcFile ) ; continue ; } StorageObjectMetadata meta ; try { meta = storageClient . getObjectMetadata ( obj . getLocation ( ) , obj . getKey ( ) ) ; } catch ( StorageProviderException spEx ) { if ( spEx . isServiceException404 ( ) ) { logger . debug ( "File returned from listing but found missing {} when getting its" + " metadata. Location={}, key={}" , obj . getLocation ( ) , obj . getKey ( ) ) ; continue ; } logger . error ( "Fetching object metadata encountered exception: {}" , spEx . getMessage ( ) ) ; throw spEx ; } String objDigest = storageClient . getDigestMetadata ( meta ) ; remoteEncrypted = MatDesc . parse ( meta . getUserMetadata ( ) . get ( storageClient . getMatdescKey ( ) ) ) != null ; InputStream fileStream = null ; String hashText = null ; final List < FileBackedOutputStream > fileBackedOutputStreams = new ArrayList < > ( ) ; try { fileStream = new FileInputStream ( localFile ) ; if ( fileMetadataMap . get ( mappedSrcFile ) . requireCompress ) { logger . debug ( "Compressing stream for digest check" ) ; InputStreamWithMetadata res = compressStreamWithGZIP ( fileStream ) ; fileStream = res . fileBackedOutputStream . asByteSource ( ) . openStream ( ) ; fileBackedOutputStreams . add ( res . fileBackedOutputStream ) ; } if ( objDigest != null ) { InputStreamWithMetadata res = computeDigest ( fileStream , false ) ; hashText = res . digest ; fileBackedOutputStreams . add ( res . fileBackedOutputStream ) ; } else if ( ! remoteEncrypted ) { hashText = DigestUtils . md5Hex ( fileStream ) ; } } finally { if ( fileStream != null ) { fileStream . close ( ) ; } for ( FileBackedOutputStream stream : fileBackedOutputStreams ) { if ( stream != null ) { try { stream . reset ( ) ; } catch ( IOException ex ) { logger . debug ( "failed to clean up temp file: {}" , ex ) ; } } } } if ( hashText == null || ( objDigest != null && ! hashText . equals ( objDigest ) ) || ( objDigest == null && ! hashText . equals ( obj . getMD5 ( ) ) ) ) { logger . debug ( "digest diff between remote store and local, will {} {}, " + "local digest: {}, remote store md5: {}" , commandType . name ( ) . toLowerCase ( ) , mappedSrcFile , hashText , obj . getMD5 ( ) ) ; continue ; } } catch ( IOException | NoSuchAlgorithmException ex ) { throw new SnowflakeSQLException ( ex , SqlState . INTERNAL_ERROR , ErrorCode . INTERNAL_ERROR . getMessageCode ( ) , "Error reading: " + localFile ) ; } logger . debug ( "digest same between remote store and local, will not upload {} {}" , commandType . name ( ) . toLowerCase ( ) , mappedSrcFile ) ; skipFile ( mappedSrcFile , objFileName ) ; } } else if ( stageInfo . getStageType ( ) == StageInfo . StageType . LOCAL_FS ) { for ( String stageFileName : stageFileNames ) { String stageFilePath = SnowflakeUtil . concatFilePathNames ( stageInfo . getLocation ( ) , stageFileName , localFSFileSep ) ; File stageFile = new File ( stageFilePath ) ; if ( ! stageFile . exists ( ) ) { continue ; } String mappedSrcFile = ( commandType == CommandType . UPLOAD ) ? destFileNameToSrcFileMap . get ( stageFileName ) : stageFileName ; String localFile = ( commandType == CommandType . UPLOAD ) ? mappedSrcFile : ( localLocation + fileMetadataMap . get ( mappedSrcFile ) . destFileName ) ; if ( ! fileMetadataMap . get ( mappedSrcFile ) . requireCompress && stageFile . length ( ) != ( new File ( localFile ) ) . length ( ) ) { logger . debug ( "Size diff between stage and local, will {} {}" , commandType . name ( ) . toLowerCase ( ) , mappedSrcFile ) ; continue ; } String localFileHashText = null ; String stageFileHashText = null ; List < FileBackedOutputStream > fileBackedOutputStreams = new ArrayList ( ) ; InputStream localFileStream = null ; try { localFileStream = new FileInputStream ( localFile ) ; if ( fileMetadataMap . get ( mappedSrcFile ) . requireCompress ) { logger . debug ( "Compressing stream for digest check" ) ; InputStreamWithMetadata res = compressStreamWithGZIP ( localFileStream ) ; fileBackedOutputStreams . add ( res . fileBackedOutputStream ) ; localFileStream = res . fileBackedOutputStream . asByteSource ( ) . openStream ( ) ; } InputStreamWithMetadata res = computeDigest ( localFileStream , false ) ; localFileHashText = res . digest ; fileBackedOutputStreams . add ( res . fileBackedOutputStream ) ; } catch ( IOException | NoSuchAlgorithmException ex ) { throw new SnowflakeSQLException ( ex , SqlState . INTERNAL_ERROR , ErrorCode . INTERNAL_ERROR . getMessageCode ( ) , "Error reading local file: " + localFile ) ; } finally { for ( FileBackedOutputStream stream : fileBackedOutputStreams ) { if ( stream != null ) { try { stream . reset ( ) ; } catch ( IOException ex ) { logger . debug ( "failed to clean up temp file: {}" , ex ) ; } } } IOUtils . closeQuietly ( localFileStream ) ; } FileBackedOutputStream fileBackedOutputStream = null ; InputStream stageFileStream = null ; try { stageFileStream = new FileInputStream ( stageFilePath ) ; InputStreamWithMetadata res = computeDigest ( stageFileStream , false ) ; stageFileHashText = res . digest ; fileBackedOutputStream = res . fileBackedOutputStream ; } catch ( IOException | NoSuchAlgorithmException ex ) { throw new SnowflakeSQLException ( ex , SqlState . INTERNAL_ERROR , ErrorCode . INTERNAL_ERROR . getMessageCode ( ) , "Error reading stage file: " + stageFilePath ) ; } finally { try { if ( fileBackedOutputStream != null ) { fileBackedOutputStream . reset ( ) ; } } catch ( IOException ex ) { logger . debug ( "failed to clean up temp file: {}" , ex ) ; } IOUtils . closeQuietly ( stageFileStream ) ; } if ( ! stageFileHashText . equals ( localFileHashText ) ) { logger . debug ( "digest diff between local and stage, will {} {}" , commandType . name ( ) . toLowerCase ( ) , mappedSrcFile ) ; continue ; } else { logger . debug ( "digest matches between local and stage, will skip {}" , mappedSrcFile ) ; skipFile ( mappedSrcFile , stageFileName ) ; } } } }
From the set of files intended to be uploaded/downloaded, derive a common
prefix and use the listObjects API to get the object summary for each
object that has the common prefix.
<p>
For each returned object, we compare the size and digest with the local file
and if they are the same, we will not upload/download the file.

@throws SnowflakeSQLException if any error occurs