docs: 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);
          }
        }
      }
    }