Only used by merge, it puts parity file's blocks and source file's blocks
together into a block array to create an INodeRaidStorage
192744
@Override
  public INodeRaidStorage convertToRaidStorage(BlockInfo[] parityBlocks, 
      RaidCodec codec, int[] checksums, BlocksMap blocksMap, short replication,
      INodeFile inode) throws IOException {
    if (codec == null) {
      throw new IOException("Codec is null");
    } else {
      return new INodeRaidStorage(codec.convertToRaidStorage(parityBlocks,
          blocks, checksums, blocksMap, replication, inode), codec);
    }
  }
360870
public void merge(Path parity, Path source, String codecId, 
      int[] checksums) throws IOException {
    dfs.merge(getPathName(parity), getPathName(source), codecId, checksums);
  }
559665
def combine(self, blocks, copy=True):
        
        if len(blocks) == 0:
            return self.make_empty()

        # FIXME: optimization potential
        indexer = np.sort(np.concatenate([b.mgr_locs.as_array
                                          for b in blocks]))
        inv_indexer = lib.get_reverse_indexer(indexer, self.shape[0])

        new_blocks = []
        for b in blocks:
            b = b.copy(deep=copy)
            b.mgr_locs = algos.take_1d(inv_indexer, b.mgr_locs.as_array,
                                       axis=0, allow_fill=False)
            new_blocks.append(b)

        axes = list(self.axes)
        axes[0] = self.items.take(indexer)

        return self.__class__(new_blocks, axes, do_integrity_check=False)
855070
private static BlockLocation[] getParityBlocks(final Path filePath,
                                          final long blockSize,
                                          final long numStripes,
                                          final RaidInfo raidInfo) 
    throws IOException {
    FileSystem parityFS = raidInfo.parityPair.getFileSystem();
    
    // get parity file metadata
    FileStatus parityFileStatus = raidInfo.parityPair.getFileStatus(); 
    long parityFileLength = parityFileStatus.getLen();

    if (parityFileLength != numStripes * raidInfo.parityBlocksPerStripe *
        blockSize) {
      throw new IOException("expected parity file of length" + 
                            (numStripes * raidInfo.parityBlocksPerStripe *
                             blockSize) +
                            " but got parity file of length " + 
                            parityFileLength);
    }

    BlockLocation[] parityBlocks = 
      parityFS.getFileBlockLocations(parityFileStatus, 0L, parityFileLength);
    
    if (parityFS instanceof DistributedFileSystem ||
        parityFS instanceof DistributedRaidFileSystem) {
      long parityBlockSize = parityFileStatus.getBlockSize();
      if (parityBlockSize != blockSize) {
        throw new IOException("file block size is " + blockSize + 
                              " but parity file block size is " + 
                              parityBlockSize);
      }
    } else if (parityFS instanceof HarFileSystem) {
      LOG.debug("HAR FS found");
    } else {
      LOG.warn("parity file system is not of a supported type");
    }
    
    return parityBlocks;
  }
186050
private boolean discardDone(INodeFileUnderConstruction pendingFile,
                              String src) throws IOException {
    Block[] blocks = pendingFile.getBlocks();
    if (blocks == null || blocks.length == 0) {
      return false;
    }
    Block last = blocks[blocks.length-1];
    if (last.getNumBytes() == 0) {
      dir.removeBlock(src, pendingFile, last);
      finalizeINodeFileUnderConstruction(src, pendingFile);
      NameNode.stateChangeLog.warn("BLOCK*" +
          " internalReleaseLease: discarded last block " + last +
          " , lease removed for " +  src);
      return true;
    }
    return false;
  }
4536
function mergeBlocks(blocks) {
  let finalBlock;
  for (const block of blocks) {
    if (!finalBlock) finalBlock = block;
    else {
      block.nodes.forEach(node => finalBlock.push(node));
    }
  }
  return finalBlock;
}
1210304
public static void create( String baseName ,
                               File dataFolder ,
                               int blockCount ,
                               int blockSize ,
                               boolean forceSync ) throws DataStoreException
    {
        if (blockCount <= 0)
            throw new DataStoreException("Block count should be > 0");
        if (blockSize <= 0)
            throw new DataStoreException("Block size should be > 0");
        
        File atFile = new File(dataFolder,baseName+AbstractBlockBasedDataStore.ALLOCATION_TABLE_SUFFIX);
        File dataFile = new File(dataFolder,baseName+AbstractBlockBasedDataStore.DATA_FILE_SUFFIX);
        if (atFile.exists())
            throw new DataStoreException("Cannot create store filesystem : "+atFile.getAbsolutePath()+" already exists");
        if (dataFile.exists())
            throw new DataStoreException("Cannot create store filesystem : "+dataFile.getAbsolutePath()+" already exists");
        
        initAllocationTable(atFile,
                            blockCount,
                            blockSize,
                            forceSync);
        initDataFile(dataFile,
                     blockCount,
                     blockSize,
                     forceSync);
    }
887909
static private boolean generateParityFile(Configuration conf, EncodingCandidate ec,
                                  int targetRepl,
                                  Progressable reporter,
                                  FileSystem inFs,
                                  Path destPathPrefix,
                                  Codec codec,
                                  long blockNum,
                                  int srcRepl,
                                  int metaRepl, 
                                  long blockSize,
                                  List<FileStatus> lfs)
                                      throws IOException, InterruptedException {
    FileStatus stat = ec.srcStat;
    Path inpath = stat.getPath();
    Path outpath =  getOriginalParityFile(destPathPrefix, inpath);
    FileSystem outFs = inFs;

    // If the parity file is already upto-date and source replication is set
    // then nothing to do.
    try {
      FileStatus stmp = outFs.getFileStatus(outpath);
      if (stmp.getModificationTime() == stat.getModificationTime() &&
          srcRepl == targetRepl) {
        LOG.info("Parity file for " + inpath + "(" + blockNum +
            ") is " + outpath + " already upto-date and " +
            "file is at target replication . Nothing more to do.");
        return false;
      }
    } catch (IOException e) {
      // ignore errors because the raid file might not exist yet.
    }

    Encoder encoder = new Encoder(conf, codec);
    encoder.verifyStore();
    StripeReader sReader = null;
    boolean parityGenerated = false;
    if (codec.isDirRaid) {
      long numStripes = (blockNum % codec.stripeLength == 0) ?
          (blockNum / codec.stripeLength) :
          ((blockNum / codec.stripeLength) + 1);
      
      sReader = new DirectoryStripeReader(conf, codec, inFs,
          ec.startStripe, ec.encodingUnit, inpath, lfs);
      parityGenerated = encoder.encodeFile(conf, inFs, outFs, outpath, 
          (short)metaRepl, numStripes, blockSize, reporter, sReader, ec);
    } else {
      FileStatus srcStat = inFs.getFileStatus(inpath);
      long srcSize = srcStat.getLen();
      long numBlocks = (srcSize % blockSize == 0) ?
                        (srcSize / blockSize) :
                        ((srcSize / blockSize) + 1);
      long numStripes = (numBlocks % codec.stripeLength == 0) ?
                        (numBlocks / codec.stripeLength) :
                        ((numBlocks / codec.stripeLength) + 1);
      sReader = new FileStripeReader(conf, blockSize, 
          codec, inFs, ec.startStripe, ec.encodingUnit, inpath, srcSize);
      parityGenerated = encoder.encodeFile(conf, inFs, outFs, outpath,
          (short)metaRepl, numStripes, blockSize, reporter, sReader, ec);
    }
    if (!parityGenerated) {
      return false;
    }

    // set the modification time of the RAID file. This is done so that the modTime of the
    // RAID file reflects that contents of the source file that it has RAIDed. This should
    // also work for files that are being appended to. This is necessary because the time on
    // on the destination namenode may not be synchronised with the timestamp of the 
    // source namenode.
    outFs.setTimes(outpath, stat.getModificationTime(), -1);

    FileStatus outstat = outFs.getFileStatus(outpath);
    FileStatus inStat = inFs.getFileStatus(inpath);
    if (stat.getModificationTime() != inStat.getModificationTime()) {
      String msg = "Source file changed mtime during raiding from " +
          stat.getModificationTime() + " to " + inStat.getModificationTime();
      throw new IOException(msg);
    }
    if (outstat.getModificationTime() != inStat.getModificationTime()) {
      String msg = "Parity file mtime " + outstat.getModificationTime() +
          " does not match source mtime " + inStat.getModificationTime();
      throw  new IOException(msg);
    }
    LOG.info("Source file " + inpath + " of size " + inStat.getLen() +
        " Parity file " + outpath + " of size " + outstat.getLen() +
        " src mtime " + stat.getModificationTime()  +
        " parity mtime " + outstat.getModificationTime());
    return true;
  }
144088
def buffer2file(self):
        
        if self.current_file is not None and self.buffer:
            self.current_file.add_block_data(self.buffered_block_length, self.buffer)
            self.buffer = []
            self.buffered_block_length = 0
1354739
private void makeBlock(String upHost, long offset, int blockSize, int chunkSize, ProgressHandler progress,
                           CompletionHandler _completionHandler, UpCancellationSignal c) {
        String path = format(Locale.ENGLISH, "/mkblk/%d", blockSize);
        try {
            file.seek(offset);
            file.read(chunkBuffer, 0, chunkSize);
        } catch (IOException e) {
            completionHandler.complete(key, ResponseInfo.fileError(e, token), null);
            return;
        }
        this.crc32 = Crc32.bytes(chunkBuffer, 0, chunkSize);
        String postUrl = String.format("%s%s", upHost, path);
        post(postUrl, chunkBuffer, 0, chunkSize, progress, _completionHandler, c);
    }
752364
public static List<DiskRange> mergeAdjacentDiskRanges(Collection<DiskRange> diskRanges, DataSize maxMergeDistance, DataSize maxReadSize)
    {
        // sort ranges by start offset
        List<DiskRange> ranges = new ArrayList<>(diskRanges);
        Collections.sort(ranges, new Comparator<DiskRange>()
        {
            @Override
            public int compare(DiskRange o1, DiskRange o2)
            {
                return Long.compare(o1.getOffset(), o2.getOffset());
            }
        });

        // merge overlapping ranges
        long maxReadSizeBytes = maxReadSize.toBytes();
        long maxMergeDistanceBytes = maxMergeDistance.toBytes();
        ImmutableList.Builder<DiskRange> result = ImmutableList.builder();
        DiskRange last = ranges.get(0);
        for (int i = 1; i < ranges.size(); i++) {
            DiskRange current = ranges.get(i);
            DiskRange merged = last.span(current);
            if (merged.getLength() <= maxReadSizeBytes && last.getEnd() + maxMergeDistanceBytes >= current.getOffset()) {
                last = merged;
            }
            else {
                result.add(last);
                last = current;
            }
        }
        result.add(last);

        return result.build();
    }
882400
public boolean processFile(Path srcPath, ParityFilePair parityPair,
      Decoder decoder, Boolean fromStripeStore, Context context) 
          throws IOException,
      InterruptedException {
    LOG.info("Processing file " + srcPath);
    Progressable progress = context;
    if (progress == null) {
      progress = RaidUtils.NULL_PROGRESSABLE;
    }
    
    DistributedFileSystem srcFs = getDFS(srcPath);
    FileStatus srcStat = srcFs.getFileStatus(srcPath);
    long blockSize = srcStat.getBlockSize();
    long srcFileSize = srcStat.getLen();
    String uriPath = srcPath.toUri().getPath();

    int numBlocksReconstructed = 0;
    List<LocatedBlockWithMetaInfo> lostBlocks = lostBlocksInFile(srcFs,
        uriPath, srcStat);
    if (lostBlocks.size() == 0) {
      LOG.warn("Couldn't find any lost blocks in file " + srcPath + 
          ", 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, 
          srcPath, lostBlockOffset, srcFs, 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;
      }

      final long blockContentsSize =
        Math.min(blockSize, srcFileSize - lostBlockOffset);
      File localBlockFile =
        File.createTempFile(lostBlock.getBlockName(), ".tmp");
      localBlockFile.deleteOnExit();

      try {
        CRC32 crc = null;
        
        if (fromStripeStore) {
          crc = decoder.recoverBlockToFileFromStripeInfo(srcFs, srcPath,
              lostBlock, localBlockFile, blockSize,
              lostBlockOffset, blockContentsSize, si, context);
        } else {
          crc = decoder.recoverBlockToFile(srcFs, srcStat,
                        parityPair.getFileSystem(),
                        parityPair.getPath(), blockSize,
                        lostBlockOffset, localBlockFile,
                        blockContentsSize, 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, srcPath, lostBlockOffset,
              LOGTYPES.OFFLINE_RECONSTRUCTION_CHECKSUM_VERIFICATION,
              srcFs, ioe, context);
          throw ioe;
        }
        // Now that we have recovered the file block locally, send it.
        computeMetadataAndSendReconstructedBlock(localBlockFile,
            lostBlock, blockContentsSize,
            lb.getLocations(),
            lb.getDataProtocolVersion(), 
            lb.getNamespaceID(), 
            progress);
        
        numBlocksReconstructed++;

      } finally {
        localBlockFile.delete();
      }
      progress.progress();
    }
    
    LOG.info("Reconstructed " + numBlocksReconstructed + " blocks in " + srcPath);
    checkLostBlocks(blocksLostChecksum, blocksLostStripe, srcPath, decoder.codec);
    return true;
  }
113111
private static <T> void mergeTwoSizeKBuffers(
      final T[] keySrc1, final int arrStart1,
      final T[] keySrc2, final int arrStart2,
      final T[] keyDst,  final int arrStart3,
      final int k, final Comparator<? super T> comparator) {
    final int arrStop1 = arrStart1 + k;
    final int arrStop2 = arrStart2 + k;

    int i1 = arrStart1;
    int i2 = arrStart2;
    int i3 = arrStart3;
    while (i1 < arrStop1 && i2 < arrStop2) {
      if (comparator.compare(keySrc2[i2], keySrc1[i1]) < 0) {
        keyDst[i3++] = keySrc2[i2++];
      } else {
        keyDst[i3++] = keySrc1[i1++];
      }
    }

    if (i1 < arrStop1) {
      System.arraycopy(keySrc1, i1, keyDst, i3, arrStop1 - i1);
    } else {
      assert i2 < arrStop2;
      System.arraycopy(keySrc1, i2, keyDst, i3, arrStop2 - i2);
    }
  }
63808
public LocatedBlock appendFile(String src, String holder, String clientMachine
      ) throws IOException {
    return appendFile(src, holder, clientMachine, BlockMetaInfoType.NONE);
  }
870045
void mergeContiguousRegions( Collection<AbstractBlock> blocks ) {

		AbstractIgnorableInstruction initialInstr = null;
		List<AbstractBlock> toRemove = new ArrayList<> ();
		StringBuilder sb = new StringBuilder();

		// We only merge comments and blank regions to reduce their number
		for( AbstractBlock block : blocks ) {
			if( initialInstr == null ) {

				if( block.getInstructionType() == AbstractBlock.COMMENT
						|| block.getInstructionType() == AbstractBlock.BLANK ) {

					AbstractIgnorableInstruction currentInstr = (AbstractIgnorableInstruction) block;
					initialInstr = currentInstr;
					sb = new StringBuilder( currentInstr.getContent());
				}

			} else if( initialInstr.getInstructionType() == block.getInstructionType()) {
				toRemove.add( block );
				sb.append( System.getProperty( "line.separator" ));
				sb.append(((AbstractIgnorableInstruction) block).getContent());

			} else {
				initialInstr.setContent( sb.toString());
				initialInstr = null;
			}
		}

		// Remove the blocks that have been merged
		blocks.removeAll( toRemove );

		// In a second time, we can reduce facets and components too
		for( AbstractBlock block : blocks ) {
			if( block.getInstructionType() == AbstractBlock.COMPONENT
					|| block.getInstructionType() == AbstractBlock.FACET )
				mergeContiguousRegions(((AbstractBlockHolder) block).getInnerBlocks());
		}
	}
349849
private void processParityHarPartBlock(FileSystem dfs, Path partFile,
      long blockOffset,
      FileStatus partFileStat,
      HarIndex harIndex,
      File localBlockFile,
      Progressable progress)
  throws IOException {
    String partName = partFile.toUri().getPath(); // Temporarily.
    partName = partName.substring(1 + partName.lastIndexOf(Path.SEPARATOR));

    OutputStream out = new FileOutputStream(localBlockFile);

    try {
      // A HAR part file block could map to several parity files. We need to
      // use all of them to recover this block.
      final long blockEnd = Math.min(blockOffset + 
          partFileStat.getBlockSize(),
          partFileStat.getLen());
      for (long offset = blockOffset; offset < blockEnd; ) {
        HarIndex.IndexEntry entry = harIndex.findEntry(partName, offset);
        if (entry == null) {
          String msg = "Lost index file has no matching index entry for " +
          partName + ":" + offset;
          LOG.warn(msg);
          throw new IOException(msg);
        }
        Path parityFile = new Path(entry.fileName);
        Encoder encoder = null;
        for (Codec codec : Codec.getCodecs()) {
          if (isParityFile(parityFile, codec)) {
            encoder = new Encoder(getConf(), codec);
          }
        }
        if (encoder == null) {
          String msg = "Could not figure out codec correctly for " + parityFile;
          LOG.warn(msg);
          throw new IOException(msg);
        }
        Path srcFile = RaidUtils.sourcePathFromParityPath(parityFile, dfs);
        if (null == srcFile) {
          String msg = "Can not find the source path for parity file: " +
                          parityFile;
          LOG.warn(msg);
          throw new IOException(msg);
        }
        FileStatus srcStat = dfs.getFileStatus(srcFile);
        if (srcStat.getModificationTime() != entry.mtime) {
          String msg = "Modification times of " + parityFile + " and " +
          srcFile + " do not match.";
          LOG.warn(msg);
          throw new IOException(msg);
        }
        long lostOffsetInParity = offset - entry.startOffset;
        LOG.info(partFile + ":" + offset + " maps to " +
            parityFile + ":" + lostOffsetInParity +
            " and will be recovered from " + srcFile);
        encoder.recoverParityBlockToStream(dfs, srcStat,
            srcStat.getBlockSize(), parityFile,
            lostOffsetInParity, out, progress);
        // Finished recovery of one parity block. Since a parity block has the
        // same size as a source block, we can move offset by source block 
        // size.
        offset += srcStat.getBlockSize();
        LOG.info("Recovered " + srcStat.getBlockSize() + " part file bytes ");
        if (offset > blockEnd) {
          String msg =
            "Recovered block spills across part file blocks. Cannot continue";
          throw new IOException(msg);
        }
        progress.progress();
      }
    } finally {
      out.close();
    }
  }
363342
public static long merge(final Collection<File> sourceFiles, final File destFile) throws UncheckedIOException {
        final byte[] buf = Objectory.createByteArrayBuffer();

        long totalCount = 0;
        OutputStream output = null;

        try {
            output = new FileOutputStream(destFile);

            InputStream input = null;
            for (File file : sourceFiles) {
                try {
                    input = new FileInputStream(file);

                    int count = 0;
                    while (EOF != (count = read(input, buf, 0, buf.length))) {
                        output.write(buf, 0, count);

                        totalCount += count;
                    }
                } finally {
                    close(input);
                }
            }

            output.flush();
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        } finally {
            Objectory.recycle(buf);

            close(output);
        }

        return totalCount;
    }