Merge parity file into source file to convert source file's storage format
from INodeRegularStorage into INodeRaidStorage.
checksums will be added to each block of the source file
return true if merge succeeds
963097
public void merge(String parity, String source, String codecId, int[] checksums) 
    throws IOException {
    if(FSNamesystem.LOG.isDebugEnabled()) {
      FSNamesystem.LOG.debug("merge " + parity + " to " + source);
    }
    if (isInSafeMode()) {
      throw new SafeModeException("merge: cannot merge " + parity + " to " 
                                  + source, safeMode);
    }
    // Verify parity and source 
    if (source == null || source.isEmpty() || parity == null || parity.isEmpty()) {
      throw new IOException(
          "merge: source file name or parity file name is empty");
    }
    // Verify checksums
    if (checksums == null || checksums.length == 0) {
      throw new IOException("merge: checksum array is empty or null");
    }
    // Verify codec
    RaidCodec codec = RaidCodec.getCodec(codecId);
    if (codec == null) {
      throw new IOException("merge: codec " + codecId + " doesn't exist");
    }
    
    INode[] sourceINodes = dir.getExistingPathINodes(source);
    INode[] parityINodes = dir.getExistingPathINodes(parity);
    writeLock();
    try {
      // write permissions for the source
      if (isPermissionEnabled) {
        if (isPermissionCheckingEnabled(sourceINodes)) {
          checkPathAccess(source, sourceINodes, FsAction.WRITE); 
        }
        if (isPermissionCheckingEnabled(parityINodes)) {
          checkPathAccess(parity, parityINodes, FsAction.READ); // read the file
          checkParentAccess(parity, parityINodes, FsAction.WRITE); // for delete
        }
      }
      INode sinode = sourceINodes[sourceINodes.length - 1];
      INode pinode = parityINodes[parityINodes.length - 1];
      
      if (sinode == null || pinode == null) {
        throw new IOException(
            "merge: source file or parity file doesn't exist");
      }
      
      if (sinode.isUnderConstruction() || pinode.isUnderConstruction()) {
        throw new IOException(
            "merge: source file or parity file is under construction");
      }
      
      if (sinode.isDirectory() || pinode.isDirectory()) {
        throw new IOException(
            "merge: source file or parity file is a directory");
      }
      
      if (sinode instanceof INodeHardLinkFile ||
          pinode instanceof INodeHardLinkFile) { 
        throw new IOException("merge: source file or parity file is hardlinked");  
      }

      INodeFile sourceINode = (INodeFile) sinode;
      INodeFile parityINode = (INodeFile) pinode;

      if (sourceINode.getStorageType() != StorageType.REGULAR_STORAGE
          || parityINode.getStorageType() != StorageType.REGULAR_STORAGE) {
        throw new IOException(
            "merge: source file or parity file doesn't support merge");
      }
      if (sourceINode.getModificationTime() != parityINode.getModificationTime()) {
        throw new IOException(
            "merge: source file and parity file doesn't have the same modification time");
      }
      if (parityINode.getReplication() != codec.parityReplication) {
        throw new IOException(
            "merge: parity file's replication doesn't match codec's parity replication");
      }
      
      BlockInfo[] sourceBlks = sourceINode.getBlocks();
      BlockInfo[] parityBlks = parityINode.getBlocks();
      
      if (sourceBlks == null || sourceBlks.length == 0) {
        throw new IOException("merge: " + source + " is empty");
      }
      if (parityBlks == null || parityBlks.length == 0) {
        throw new IOException("merge: " + parity + " is empty");
      }
      if (checksums.length != sourceBlks.length) {
        throw new IOException("merge: checksum length " + checksums.length +
            " doesn't match number of source blocks " + sourceBlks.length);
      }
      int expectedParityBlocks = codec.getNumStripes(sourceBlks.length) 
          * codec.numParityBlocks;
      if (expectedParityBlocks != parityBlks.length) {
        throw new IOException("merge: expect parity blocks " + 
            expectedParityBlocks + " doesn't match number of parity blocks " + 
            parityBlks.length);
      }

      if(NameNode.stateChangeLog.isDebugEnabled()) {
        NameNode.stateChangeLog.debug("DIR* NameSystem.merge: " +
            parity + " to " + source);
      }
      
      dir.mergeInternal(parityINodes, sourceINodes, parity, source, codec,
                        checksums);
    } finally {
      writeUnlock();
    }
    getEditLog().logSync();

    if (auditLog.isInfoEnabled()) {
      logAuditEvent(getCurrentUGI(),
                    Server.getRemoteIp(),
                    "merge", parity, source, 
                    getLastINode(sourceINodes));
    }
  }
1225075
private INodeFileUnderConstruction checkLease(String src, String holder)
    throws IOException {
    INodeFile file = dir.getFileINode(src);
    return checkLease(src, holder, file);
  }
1170098
def validate_files(directory, files_to_merge):
    
    for file_path in files_to_merge:
        pathname = directory.joinpath(file_path)
        if not pathname.exists():
            raise Exception("I18N: Cannot generate because file not found: {0}".format(pathname))
        # clean sources
        clean_pofile(pathname)
1090728
public boolean reportChecksumFailure(Path p, FSDataInputStream in,
                                       long inPos,
                                       FSDataInputStream sums, long sumsPos) {
    try {
      // canonicalize f
      File f = ((RawLocalFileSystem)fs).pathToFile(p).getCanonicalFile();
      
      // find highest writable parent dir of f on the same device
      String device = new DF(f, getConf()).getMount();
      File parent = f.getParentFile();
      File dir = null;
      while (parent!=null && parent.canWrite() && parent.toString().startsWith(device)) {
        dir = parent;
        parent = parent.getParentFile();
      }

      if (dir==null) {
        throw new IOException(
                              "not able to find the highest writable parent dir");
      }
        
      // move the file there
      File badDir = new File(dir, "bad_files");
      if (!badDir.mkdirs()) {
        if (!badDir.isDirectory()) {
          throw new IOException("Mkdirs failed to create " + badDir.toString());
        }
      }
      String suffix = "." + rand.nextInt();
      File badFile = new File(badDir, f.getName()+suffix);
      LOG.warn("Moving bad file " + f + " to " + badFile);
      in.close();                               // close it first
      f.renameTo(badFile);                      // rename it

      // move checksum file too
      File checkFile = ((RawLocalFileSystem)fs).pathToFile(getChecksumFile(p));
      checkFile.renameTo(new File(badDir, checkFile.getName()+suffix));

    } catch (IOException e) {
      LOG.warn("Error moving bad file " + p + ": " + e);
    }
    return false;
  }
193729
public static boolean raidedByOtherHighPriCodec(Configuration conf,
        FileStatus stat, Codec codec) throws IOException {
    
    for (Codec tcodec : Codec.getCodecs()) {
      if (tcodec.priority > codec.priority) {
        if (stat.isDir() && !tcodec.isDirRaid) {
          // A directory could not be raided by a file level codec.
          continue;
        }
        
        // check if high priority parity file exists.
        if (ParityFilePair.parityExists(stat, tcodec, conf)) {
          InjectionHandler.processEvent(InjectionEvent.RAID_ENCODING_SKIP_PATH);
          return true;
        }
      }
    }
    
    return false;
  }
956536
public static boolean parityExists(FileStatus src, Codec codec,
      Configuration conf) throws IOException {
    return ParityFilePair.getParityFile(codec, src, conf) != null;
  }
414470
boolean reconstructFile(Path srcPath, Context context)
      throws IOException, InterruptedException {
    Progressable progress = context;
    if (progress == null) {
      progress = RaidUtils.NULL_PROGRESSABLE;
    }
    
    FileSystem fs = srcPath.getFileSystem(getConf());
    FileStatus srcStat = null;
    try {
      srcStat = fs.getFileStatus(srcPath);
    } catch (FileNotFoundException ex) {
      return false;
    }

    if (RaidNode.isParityHarPartFile(srcPath)) {
      return processParityHarPartFile(srcPath, progress);
    }

    // Reconstruct parity file
    for (Codec codec : Codec.getCodecs()) {
      if (isParityFile(srcPath, codec)) {
        Decoder decoder = new Decoder(getConf(), codec);
        decoder.connectToStore(srcPath);
        return processParityFile(srcPath, 
            decoder, context);
      }
    }

    // Reconstruct source file without connecting to stripe store 
    for (Codec codec : Codec.getCodecs()) {
      ParityFilePair ppair = ParityFilePair.getParityFile(
          codec, srcStat, getConf());
      if (ppair != null) {
        Decoder decoder = new Decoder(getConf(), codec);
        decoder.connectToStore(srcPath);
        return processFile(srcPath, ppair, decoder, false, context);
      }
    }
    // Reconstruct source file through stripe store
    for (Codec codec : Codec.getCodecs()) {
      if (!codec.isDirRaid) {
        continue;
      }
      try {
        // try to fix through the stripe store.
        Decoder decoder = new Decoder(getConf(), codec);
        decoder.connectToStore(srcPath);
        if (processFile(srcPath, null, decoder, true, context)) {
          return true;
        }
      } catch (Exception ex) {
        LogUtils.logRaidReconstructionMetrics(LOGRESULTS.FAILURE, 0,
            codec, srcPath, -1, LOGTYPES.OFFLINE_RECONSTRUCTION_USE_STRIPE,
            fs, ex, context);
      }
    }
    
    return false;    
  }
1372825
synchronized boolean add(
                           BlockInfo blockInfo,
                           int curReplicas, 
                           int decomissionedReplicas,
                           int expectedReplicas) {
    if(curReplicas<0 || expectedReplicas <= curReplicas) {
      return false;
    }
    int priLevel = getPriority(blockInfo, curReplicas, decomissionedReplicas,
                               expectedReplicas);
    INodeFile fileINode = blockInfo.getINode();
    if (priLevel == QUEUE_WITH_CORRUPT_BLOCKS && fileINode != null &&
        fileINode.getStorageType().equals(StorageType.RAID_STORAGE)) {
      RaidCodec codec = ((INodeRaidStorage) fileINode.getStorage()).getCodec();
      return raidQueue.add(blockInfo, codec);
    }
    
    if(priLevel != LEVEL && priorityQueues.get(priLevel).add(blockInfo)) {
      if (NameNode.stateChangeLog.isDebugEnabled()) {
        NameNode.stateChangeLog.debug(
                                      "BLOCK* NameSystem.UnderReplicationBlock.add:"
                                      + blockInfo
                                      + " has only "+curReplicas
                                      + " replicas and need " + expectedReplicas
                                      + " replicas so is added to neededReplications"
                                      + " at priority level " + priLevel);
      }
      return true;
    }
    return false;
  }
959526
public boolean reportChecksumFailure(Path f, 
    FSDataInputStream in, long inPos, 
    FSDataInputStream sums, long sumsPos) {

    LocatedBlock lblocks[] = new LocatedBlock[2];

    // Find block in data stream.
    DFSClient.DFSDataInputStream dfsIn = (DFSClient.DFSDataInputStream) in;
    Block dataBlock = dfsIn.getCurrentBlock();
    if (dataBlock == null) {
      LOG.error("Error: Current block in data stream is null! ");
      return false;
    }
    DatanodeInfo[] dataNode = {dfsIn.getCurrentDatanode()}; 
    lblocks[0] = new LocatedBlock(dataBlock, dataNode);
    LOG.info("Found checksum error in data stream at block="
        + dataBlock + " on datanode="
        + dataNode[0].getName());

    // Find block in checksum stream
    DFSClient.DFSDataInputStream dfsSums = (DFSClient.DFSDataInputStream) sums;
    Block sumsBlock = dfsSums.getCurrentBlock();
    if (sumsBlock == null) {
      LOG.error("Error: Current block in checksum stream is null! ");
      return false;
    }
    DatanodeInfo[] sumsNode = {dfsSums.getCurrentDatanode()}; 
    lblocks[1] = new LocatedBlock(sumsBlock, sumsNode);
    LOG.info("Found checksum error in checksum stream at block="
        + sumsBlock + " on datanode="
        + sumsNode[0].getName());

    // Ask client to delete blocks.
    dfs.reportChecksumFailure(f.toString(), lblocks);

    return true;
  }
597642
boolean processParityFile(Path parityPath, Decoder decoder, 
      Context context)
  throws IOException, InterruptedException {
    LOG.info("Processing parity file " + parityPath);
    
    Progressable progress = context;
    if (progress == null) {
      progress = RaidUtils.NULL_PROGRESSABLE;
    }
    DistributedFileSystem parityFs = getDFS(parityPath);
    Path srcPath = RaidUtils.sourcePathFromParityPath(parityPath, parityFs);
    if (srcPath == null) {
      LOG.warn("Could not get regular file corresponding to parity file " +  
          parityPath + ", ignoring...");
      return false;
    }
    
    DistributedFileSystem srcFs = getDFS(srcPath);
    FileStatus parityStat = parityFs.getFileStatus(parityPath);
    long blockSize = parityStat.getBlockSize();
    FileStatus srcStat = srcFs.getFileStatus(srcPath);

    // Check timestamp.
    if (srcStat.getModificationTime() != parityStat.getModificationTime()) {
      LOG.warn("Mismatching timestamp for " + srcPath + " and " + parityPath + 
          ", ignoring...");
      return false;
    }

    String uriPath = parityPath.toUri().getPath();
    int numBlocksReconstructed = 0;
    List<LocatedBlockWithMetaInfo> lostBlocks = 
      lostBlocksInFile(parityFs, uriPath, parityStat);
    if (lostBlocks.size() == 0) {
      LOG.warn("Couldn't find any lost blocks in parity file " + parityPath + 
          ", ignoring...");
      return false;
    }
    List<Block> blocksLostChecksum = new ArrayList<Block>();
    List<Block> blocksLostStripe = new ArrayList<Block>();
    
    for (LocatedBlockWithMetaInfo lb: lostBlocks) {
      Block lostBlock = lb.getBlock();
      long lostBlockOffset = lb.getStartOffset();
      LOG.info("Found lost block " + lostBlock +
          ", offset " + lostBlockOffset);
      
      Long oldCRC = decoder.retrieveChecksum(lostBlock,
          parityPath, lostBlockOffset, parityFs, context);
      if (abortReconstruction(oldCRC, decoder)) {
        blocksLostChecksum.add(lostBlock);
        continue;
      }
      StripeInfo si = decoder.retrieveStripe(lostBlock,
          srcPath, lostBlockOffset, srcFs, context, false);
      if (si == null && decoder.stripeStore != null) {
        blocksLostStripe.add(lostBlock);
        continue;
      }
      
      File localBlockFile =
        File.createTempFile(lostBlock.getBlockName(), ".tmp");
      localBlockFile.deleteOnExit();

      try {
        CRC32 crc = decoder.recoverParityBlockToFile(srcFs, srcStat, parityFs,
            parityPath, blockSize, lostBlockOffset, localBlockFile, si,
            context);
        LOG.info("Recovered crc: " + ((crc == null)?null: crc.getValue()) +
            " expected crc:" + oldCRC);
        if (crc != null && oldCRC != null &&
            crc.getValue() != oldCRC) {
          // checksum doesn't match, it's dangerous to send it
          IOException ioe = new IOException("Block " + lostBlock.toString()
              + " new checksum " + crc.getValue()
              + " doesn't match the old one " + oldCRC);
          LogUtils.logRaidReconstructionMetrics(LOGRESULTS.FAILURE, 0,
              decoder.codec, parityPath, lostBlockOffset,
              LOGTYPES.OFFLINE_RECONSTRUCTION_CHECKSUM_VERIFICATION,
              parityFs, ioe, context);
          throw ioe;
        }
        // Now that we have recovered the parity file block locally, send it.
        computeMetadataAndSendReconstructedBlock(
            localBlockFile, 
            lostBlock, blockSize, lb.getLocations(),
            lb.getDataProtocolVersion(), lb.getNamespaceID(),
            progress);

        numBlocksReconstructed++;
      } finally {
        localBlockFile.delete();
      }
      progress.progress();
    }
    
    LOG.info("Reconstructed " + numBlocksReconstructed + " blocks in " + parityPath);
    checkLostBlocks(blocksLostChecksum, blocksLostStripe, parityPath, decoder.codec);
    return true;
  }
321888
def append_checksum(self, f, checksum):
        
        align_file_position(f, 16)
        f.write(struct.pack(b'B', checksum))
557122
def _consolidate_repo_sources(sources):
    '''
    
    '''
    if not isinstance(sources, sourceslist.SourcesList):
        raise TypeError(
            '\'{0}\' not a \'{1}\''.format(
                type(sources),
                sourceslist.SourcesList
            )
        )

    consolidated = {}
    delete_files = set()
    base_file = sourceslist.SourceEntry('').file

    repos = [s for s in sources.list if not s.invalid]

    for repo in repos:
        repo.uri = repo.uri.rstrip('/')
        # future lint: disable=blacklisted-function
        key = str((getattr(repo, 'architectures', []),
                   repo.disabled, repo.type, repo.uri, repo.dist))
        # future lint: enable=blacklisted-function
        if key in consolidated:
            combined = consolidated[key]
            combined_comps = set(repo.comps).union(set(combined.comps))
            consolidated[key].comps = list(combined_comps)
        else:
            consolidated[key] = sourceslist.SourceEntry(salt.utils.pkg.deb.strip_uri(repo.line))

        if repo.file != base_file:
            delete_files.add(repo.file)

    sources.list = list(consolidated.values())
    sources.save()
    for file_ in delete_files:
        try:
            os.remove(file_)
        except OSError:
            pass
    return sources
939758
private void verifyChunkedSums(
      byte[] data, int dataOff, int dataLen,
      byte[] checksums, int checksumsOff, String fileName,
      long basePos) throws ChecksumException {
    
    int remaining = dataLen;
    int dataPos = 0;
    while (remaining > 0) {
      int n = Math.min(remaining, bytesPerChecksum);
      
      summer.reset();
      summer.update(data, dataOff + dataPos, n);
      dataPos += n;
      remaining -= n;
      
      int calculated = (int)summer.getValue();
      int stored = (checksums[checksumsOff] << 24 & 0xff000000) |
        (checksums[checksumsOff + 1] << 16 & 0xff0000) |
        (checksums[checksumsOff + 2] << 8 & 0xff00) |
        checksums[checksumsOff + 3] & 0xff;
      checksumsOff += 4;
      if (calculated != stored) {
        long errPos = basePos + dataPos - n;
        throw new ChecksumException(
            "Checksum error: "+ fileName + " at "+ errPos +
            " exp: " + stored + " got: " + calculated, errPos);
      }
    }
  }
88068
def delayed_bam_merge(data):
    
    if data.get("combine"):
        assert len(data["combine"].keys()) == 1
        file_key = list(data["combine"].keys())[0]
        extras = []
        for x in data["combine"][file_key].get("extras", []):
            if isinstance(x, (list, tuple)):
                extras.extend(x)
            else:
                extras.append(x)
        if file_key in data:
            extras.append(data[file_key])
        in_files = sorted(list(set(extras)))
        out_file = tz.get_in(["combine", file_key, "out"], data, _merge_out_from_infiles(in_files))
        sup_exts = data.get(file_key + "_plus", {}).keys()
        for ext in list(sup_exts) + [""]:
            merged_file = None
            if os.path.exists(utils.append_stem(out_file, "-" + ext)):
                cur_out_file, cur_in_files = out_file, []
            if ext:
                cur_in_files = list(filter(os.path.exists, (utils.append_stem(f, "-" + ext) for f in in_files)))
                cur_out_file = utils.append_stem(out_file, "-" + ext) if len(cur_in_files) > 0 else None
            else:
                cur_in_files, cur_out_file = in_files, out_file
            if cur_out_file:
                config = copy.deepcopy(data["config"])
                if len(cur_in_files) > 0:
                    merged_file = merge_bam_files(cur_in_files, os.path.dirname(cur_out_file), data,
                                                  out_file=cur_out_file)
                else:
                    assert os.path.exists(cur_out_file)
                    merged_file = cur_out_file
            if merged_file:
                if ext:
                    data[file_key + "_plus"][ext] = merged_file
                else:
                    data[file_key] = merged_file
        data.pop("region", None)
        data.pop("combine", None)
    return [[data]]
498778
def merge(blocks):
    

    current_block = blocks[sorted(blocks.keys())[0]]

    compressed_data = []
    eof = False

    while not eof:
        data_size_to_append = None
        next_block = None

        i = 0
        while i < len(current_block.data) - 1:
            current_byte = current_block.data[i]
            next_byte = current_block.data[i + 1]

            if current_byte == RLE_BYTE:
                if next_byte == RLE_BYTE:
                    i += 2
                else:
                    i += 3
            elif current_byte == SPECIAL_BYTE:
                if next_byte in SPECIAL_DEFAULTS:
                    i += 3
                elif next_byte == SPECIAL_BYTE:
                    i += 2
                else:
                    data_size_to_append = i

                    # hit end of file
                    if next_byte == EOF_BYTE:
                        eof = True
                    else:
                        next_block = blocks[next_byte]

                    break
            else:
                i += 1

        assert data_size_to_append is not None, "Ran off the end of a "\
            "block without encountering a block switch or EOF"

        compressed_data.extend(current_block.data[0:data_size_to_append])

        if not eof:
            assert next_block is not None, "Switched blocks, but did " \
                "not provide the next block to switch to"

            current_block = next_block

    return compressed_data
200565
def _merge_out_from_infiles(in_files):
    
    fname = os.path.commonprefix([os.path.basename(f) for f in in_files])
    while fname.endswith(("-", "_", ".")):
        fname = fname[:-1]
    ext = os.path.splitext(in_files[0])[-1]
    dirname = os.path.dirname(in_files[0])
    while dirname.endswith(("split", "merge")):
        dirname = os.path.dirname(dirname)
    return os.path.join(dirname, "%s%s" % (fname, ext))
1033756
public void raidFile(INode sourceINodes[], String source, 
      RaidCodec codec, short expectedSourceRepl, Block[] parityBlocks)
          throws IOException {
    waitForReady();
    long now = FSNamesystem.now();
    unprotectedRaidFile(sourceINodes, source, codec, expectedSourceRepl,
        parityBlocks, now);
    fsImage.getEditLog().logRaidFile(source, codec.id, expectedSourceRepl,
        now);
    
  }