HBase源码系列之HFile

long org.apache.hadoop.hbase.regionserver.StoreFlusher.performFlush(InternalScanner scanner, CellSink sink, long smallestReadPoint) throws IOException

public void (final KeyValue kv) throws IOException {
  
  appendGeneralBloomfilter(kv);
  appendDeleteFamilyBloomFilter(kv);
  //这行是重点
  writer.append(kv);
  //这行先不管，处理时间戳
  trackTimestamps(kv);
}

//检查key是否有问题，是否按顺序（memstore使用ConcurrentSkipListMap存储，应该不会有此问题）。
//并且返回key是否重复

boolean dupKey = checkKey(key, koffset, klength);
checkValue(value, voffset, vlength);
//如果不重复，则不检查边界，答案不能，因为如果有重复，不会检查边界更不会新建一个block。***问题5***

if (!dupKey) {
  //此出会检查block的大小，并且有一处需要注意，在里面的代码中有一些记录block信息的，这个以后会有用。
  //此处会写出chunk，处理readyChunks
  checkBlockBoundary();
}

byte[] indexKey = comparator.calcIndexKey(lastKeyOfPreviousBlock, firstKeyInBlock);
dataBlockIndexWriter.addEntry(indexKey, lastDataBlockOffset, onDiskSize);

//在此代码中会判断key的个数，如果key的个数达到了一定程度就新建一个chunk，放入readyChunks（这个会在checkBlockBoundary中处理），此出会写bf。***问题2***
enqueueReadyChunk(false);
... 这种是处理chunk被写出的时候的操作。重置一些值 ...
//真正的添加到bf中
chunk.add(bloomKey, keyOffset, keyLength);

ReadyChunk readyChunk = new ReadyChunk();
readyChunk.chunkId = numChunks - 1;
readyChunk.chunk = chunk;
readyChunk.firstKey = firstKeyInChunk; 
readyChunks.add(readyChunk);

//此处组织了block，将加入到此HFile的chunk生成树的结构。

long rootIndexOffset = dataBlockIndexWriter.writeIndexBlocks(outputStream);

private void writeInlineBlocks(boolean closing) throws IOException {
  //inlineBlockWriters 应该就3个，两个bf和一个block（待确定）
  for (InlineBlockWriter ibw : inlineBlockWriters) {
    while (ibw.shouldWriteBlock(closing)) {
      long offset = outputStream.getPos();
      boolean cacheThisBlock = ibw.getCacheOnWrite();
      ibw.writeInlineBlock(fsBlockWriter.startWriting(
          ibw.getInlineBlockType()));
      fsBlockWriter.writeHeaderAndData(outputStream);
      //此处添加leaf index block
      ibw.blockWritten(offset, fsBlockWriter.getOnDiskSizeWithHeader(),
          fsBlockWriter.getUncompressedSizeWithoutHeader());
      totalUncompressedBytes += fsBlockWriter.getUncompressedSizeWithHeader();
      if (cacheThisBlock) {
        doCacheOnWrite(offset);
      }
    }
  }
}

public boolean shouldWriteBlock(boolean closing) {
  enqueueReadyChunk(closing);
  //readyChunks中保存的是chunk，也就是lead index block
  return !readyChunks.isEmpty();
}

bloomBlockIndexWriter.writeSingleLevelIndex(out, "Bloom filter");

List<StoreFileScanner> sfScanners = StoreFileScanner.getScannersForStoreFiles(storeFilesToScan, cacheBlocks, usePread, isCompaction, false, matcher, readPt);

//此方法会调用Open方法
StoreFile.Reader r = file.createReader(canUseDrop);

   if (this.reader != null) {
     throw new IllegalAccessError("Already open");
   }
   // Open the StoreFile.Reader
   this.reader = fileInfo.open(this.fs, this.cacheConf, canUseDropBehind);
   // Load up indices and fileinfo. This also loads Bloom filter type.
   metadataMap = Collections.unmodifiableMap(this.reader.loadFileInfo());
   // Read in our metadata.
   byte [] b = metadataMap.get(MAX_SEQ_ID_KEY);
   if (b != null) {
     // By convention, if halfhfile, top half has a sequence number > bottom
     // half. Thats why we add one in below. Its done for case the two halves
     // are ever merged back together --rare.  Without it, on open of store,
     // since store files are distinguished by sequence id, the one half would
     // subsume the other.
     this.sequenceid = Bytes.toLong(b);
     if (fileInfo.isTopReference()) {
       this.sequenceid += 1;
     }
   }
   if (isBulkLoadResult()){
     // generate the sequenceId from the fileName
     // fileName is of the form <randomName>_SeqId_<id-when-loaded>_
     String fileName = this.getPath().getName();
     // Use lastIndexOf() to get the last, most recent bulk load seqId.
     int startPos = fileName.lastIndexOf("SeqId_");
     if (startPos != -1) {
       this.sequenceid = Long.parseLong(fileName.substring(startPos + 6,
           fileName.indexOf('_', startPos + 6)));
       // Handle reference files as done above.
       if (fileInfo.isTopReference()) {
         this.sequenceid += 1;
       }
     }
     this.reader.setBulkLoaded(true);
   }
   this.reader.setSequenceID(this.sequenceid);
   b = metadataMap.get(HFileWriterV2.MAX_MEMSTORE_TS_KEY);
   if (b != null) {
     this.maxMemstoreTS = Bytes.toLong(b);
   }
   b = metadataMap.get(MAJOR_COMPACTION_KEY);
   if (b != null) {
     boolean mc = Bytes.toBoolean(b);
     if (this.majorCompaction == null) {
       this.majorCompaction = new AtomicBoolean(mc);
     } else {
       this.majorCompaction.set(mc);
     }
   } else {
     // Presume it is not major compacted if it doesn't explicity say so
     // HFileOutputFormat explicitly sets the major compacted key.
     this.majorCompaction = new AtomicBoolean(false);
   }
   b = metadataMap.get(EXCLUDE_FROM_MINOR_COMPACTION_KEY);
   this.excludeFromMinorCompaction = (b != null && Bytes.toBoolean(b));
//此出会读取bloom filter
   BloomType hfileBloomType = reader.getBloomFilterType();
   if (cfBloomType != BloomType.NONE) {
     reader.loadBloomfilter(BlockType.GENERAL_BLOOM_META);
     if (hfileBloomType != cfBloomType) {
       LOG.info("HFile Bloom filter type for "
           + reader.getHFileReader().getName() + ": " + hfileBloomType
           + ", but " + cfBloomType + " specified in column family "
           + "configuration");
     }
   } else if (hfileBloomType != BloomType.NONE) {
     LOG.info("Bloom filter turned off by CF config for "
         + reader.getHFileReader().getName());
   }
   // load delete family bloom filter
   reader.loadBloomfilter(BlockType.DELETE_FAMILY_BLOOM_META);
   try {
     this.reader.timeRange = TimeRangeTracker.getTimeRange(metadataMap.get(TIMERANGE_KEY));
   } catch (IllegalArgumentException e) {
     LOG.error("Error reading timestamp range data from meta -- " +
         "proceeding without", e);
     this.reader.timeRange = null;
   }
   return this.reader;

//此处使用bloomfilter过滤。在此方法中会调用bloomFilter.contains，在此contains会先使用block index 判断。
reader.passesBloomFilter(scan, columns)

   //判断读取哪个block，rootBlockContaingKey里的blockKeys为chunk的个数。
//index是从bloommeta中读取，DataInput bloomMeta = reader.getGeneralBloomFilterMetadata(); 代码获取。

int block = index.rootBlockContainingKey(key, keyOffset, keyLength);
   if (block < 0) {
     result = false; // This key is not in the file.
   } else {
     HFileBlock bloomBlock;
     try {
       // We cache the block and use a positional read.
       //读取那个chunk的bf
       bloomBlock = reader.readBlock(index.getRootBlockOffset(block),
           index.getRootBlockDataSize(block), true, true, false, true,
           BlockType.BLOOM_CHUNK);
     } catch (IOException ex) {
       // The Bloom filter is broken, turn it off.
       throw new IllegalArgumentException(
           "Failed to load Bloom block for key "
               + Bytes.toStringBinary(key, keyOffset, keyLength), ex);
     }
     ByteBuffer bloomBuf = bloomBlock.getBufferReadOnly();
     result = ByteBloomFilter.contains(key, keyOffset, keyLength,
         bloomBuf.array(), bloomBuf.arrayOffset() + bloomBlock.headerSize(),
         bloomBlock.getUncompressedSizeWithoutHeader(), hash, hashCount);
   }

BlockWithScanInfo org.apache.hadoop.hbase.io.hfile.HFileBlockIndex.BlockIndexReader.loadDataBlockWithScanInfo(byte[] key, int keyOffset, int keyLength, HFileBlock currentBlock, boolean cacheBlocks, boolean pread, boolean isCompaction) throws IOException