001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.io.hfile;
019
020import static org.apache.hadoop.hbase.io.hfile.BlockCompressedSizePredicator.MAX_BLOCK_SIZE_UNCOMPRESSED;
021import static org.apache.hadoop.hbase.regionserver.CustomTieringMultiFileWriter.CUSTOM_TIERING_TIME_RANGE;
022import static org.apache.hadoop.hbase.regionserver.HStoreFile.EARLIEST_PUT_TS;
023import static org.apache.hadoop.hbase.regionserver.HStoreFile.TIMERANGE_KEY;
024
025import java.io.DataOutput;
026import java.io.DataOutputStream;
027import java.io.IOException;
028import java.net.InetSocketAddress;
029import java.nio.ByteBuffer;
030import java.util.ArrayList;
031import java.util.List;
032import java.util.Optional;
033import java.util.function.Supplier;
034import org.apache.hadoop.conf.Configuration;
035import org.apache.hadoop.fs.FSDataOutputStream;
036import org.apache.hadoop.fs.FileSystem;
037import org.apache.hadoop.fs.Path;
038import org.apache.hadoop.fs.permission.FsPermission;
039import org.apache.hadoop.hbase.ByteBufferExtendedCell;
040import org.apache.hadoop.hbase.Cell;
041import org.apache.hadoop.hbase.CellComparator;
042import org.apache.hadoop.hbase.CellUtil;
043import org.apache.hadoop.hbase.ExtendedCell;
044import org.apache.hadoop.hbase.HConstants;
045import org.apache.hadoop.hbase.KeyValue;
046import org.apache.hadoop.hbase.KeyValueUtil;
047import org.apache.hadoop.hbase.MetaCellComparator;
048import org.apache.hadoop.hbase.PrivateCellUtil;
049import org.apache.hadoop.hbase.io.compress.Compression;
050import org.apache.hadoop.hbase.io.crypto.Encryption;
051import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
052import org.apache.hadoop.hbase.io.encoding.IndexBlockEncoding;
053import org.apache.hadoop.hbase.io.hfile.HFileBlock.BlockWritable;
054import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
055import org.apache.hadoop.hbase.security.EncryptionUtil;
056import org.apache.hadoop.hbase.security.User;
057import org.apache.hadoop.hbase.util.BloomFilterWriter;
058import org.apache.hadoop.hbase.util.ByteBufferUtils;
059import org.apache.hadoop.hbase.util.Bytes;
060import org.apache.hadoop.hbase.util.CommonFSUtils;
061import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
062import org.apache.hadoop.hbase.util.FSUtils;
063import org.apache.hadoop.io.Writable;
064import org.apache.yetus.audience.InterfaceAudience;
065import org.slf4j.Logger;
066import org.slf4j.LoggerFactory;
067
068/**
069 * Common functionality needed by all versions of {@link HFile} writers.
070 */
071@InterfaceAudience.Private
072public class HFileWriterImpl implements HFile.Writer {
073  private static final Logger LOG = LoggerFactory.getLogger(HFileWriterImpl.class);
074
075  private static final long UNSET = -1;
076
077  /** if this feature is enabled, preCalculate encoded data size before real encoding happens */
078  public static final String UNIFIED_ENCODED_BLOCKSIZE_RATIO =
079    "hbase.writer.unified.encoded.blocksize.ratio";
080
081  /** Block size limit after encoding, used to unify encoded block Cache entry size */
082  private final int encodedBlockSizeLimit;
083
084  /** The Cell previously appended. Becomes the last cell in the file. */
085  protected ExtendedCell lastCell = null;
086
087  /** FileSystem stream to write into. */
088  protected FSDataOutputStream outputStream;
089
090  /** True if we opened the <code>outputStream</code> (and so will close it). */
091  protected final boolean closeOutputStream;
092
093  /** A "file info" block: a key-value map of file-wide metadata. */
094  protected HFileInfo fileInfo = new HFileInfo();
095
096  /** Total # of key/value entries, i.e. how many times add() was called. */
097  protected long entryCount = 0;
098
099  /** Used for calculating the average key length. */
100  protected long totalKeyLength = 0;
101
102  /** Used for calculating the average value length. */
103  protected long totalValueLength = 0;
104
105  /** Len of the biggest cell. */
106  protected long lenOfBiggestCell = 0;
107  /** Key of the biggest cell. */
108  protected byte[] keyOfBiggestCell;
109
110  /** Total uncompressed bytes, maybe calculate a compression ratio later. */
111  protected long totalUncompressedBytes = 0;
112
113  /** Meta block names. */
114  protected List<byte[]> metaNames = new ArrayList<>();
115
116  /** {@link Writable}s representing meta block data. */
117  protected List<Writable> metaData = new ArrayList<>();
118
119  /**
120   * First cell in a block. This reference should be short-lived since we write hfiles in a burst.
121   */
122  protected ExtendedCell firstCellInBlock = null;
123
124  /** May be null if we were passed a stream. */
125  protected final Path path;
126
127  protected final Configuration conf;
128
129  /** Cache configuration for caching data on write. */
130  protected final CacheConfig cacheConf;
131
132  public void setTimeRangeTrackerForTiering(Supplier<TimeRangeTracker> timeRangeTrackerForTiering) {
133    this.timeRangeTrackerForTiering = timeRangeTrackerForTiering;
134  }
135
136  private Supplier<TimeRangeTracker> timeRangeTrackerForTiering;
137
138  /**
139   * Name for this object used when logging or in toString. Is either the result of a toString on
140   * stream or else name of passed file Path.
141   */
142  protected final String name;
143
144  /**
145   * The data block encoding which will be used. {@link NoOpDataBlockEncoder#INSTANCE} if there is
146   * no encoding.
147   */
148  protected final HFileDataBlockEncoder blockEncoder;
149
150  protected final HFileIndexBlockEncoder indexBlockEncoder;
151
152  protected final HFileContext hFileContext;
153
154  private int maxTagsLength = 0;
155
156  /** KeyValue version in FileInfo */
157  public static final byte[] KEY_VALUE_VERSION = Bytes.toBytes("KEY_VALUE_VERSION");
158
159  /** Version for KeyValue which includes memstore timestamp */
160  public static final int KEY_VALUE_VER_WITH_MEMSTORE = 1;
161
162  /** Inline block writers for multi-level block index and compound Blooms. */
163  private List<InlineBlockWriter> inlineBlockWriters = new ArrayList<>();
164
165  /** block writer */
166  protected HFileBlock.Writer blockWriter;
167
168  private HFileBlockIndex.BlockIndexWriter dataBlockIndexWriter;
169  private HFileBlockIndex.BlockIndexWriter metaBlockIndexWriter;
170
171  /** The offset of the first data block or -1 if the file is empty. */
172  private long firstDataBlockOffset = UNSET;
173
174  /** The offset of the last data block or 0 if the file is empty. */
175  protected long lastDataBlockOffset = UNSET;
176
177  /**
178   * The last(stop) Cell of the previous data block. This reference should be short-lived since we
179   * write hfiles in a burst.
180   */
181  private ExtendedCell lastCellOfPreviousBlock = null;
182
183  /** Additional data items to be written to the "load-on-open" section. */
184  private List<BlockWritable> additionalLoadOnOpenData = new ArrayList<>();
185
186  protected long maxMemstoreTS = 0;
187
188  private final TimeRangeTracker timeRangeTracker;
189  private long earliestPutTs = HConstants.LATEST_TIMESTAMP;
190
191  private String regionName;
192
193  private String familyName;
194
195  public HFileWriterImpl(final Configuration conf, CacheConfig cacheConf, Path path,
196    FSDataOutputStream outputStream, HFileContext fileContext) {
197    this.outputStream = outputStream;
198    this.path = path;
199    if (path != null) {
200      this.name = path.getName();
201      this.regionName = path.getParent().getParent().getName();
202      this.familyName = path.getParent().getName();
203    } else {
204      this.name = outputStream.toString();
205    }
206    this.hFileContext = fileContext;
207    // TODO: Move this back to upper layer
208    this.timeRangeTracker = TimeRangeTracker.create(TimeRangeTracker.Type.NON_SYNC);
209    this.timeRangeTrackerForTiering = () -> this.timeRangeTracker;
210    DataBlockEncoding encoding = hFileContext.getDataBlockEncoding();
211    if (encoding != DataBlockEncoding.NONE) {
212      this.blockEncoder = new HFileDataBlockEncoderImpl(encoding);
213    } else {
214      this.blockEncoder = NoOpDataBlockEncoder.INSTANCE;
215    }
216    IndexBlockEncoding indexBlockEncoding = hFileContext.getIndexBlockEncoding();
217    if (indexBlockEncoding != IndexBlockEncoding.NONE) {
218      this.indexBlockEncoder = new HFileIndexBlockEncoderImpl(indexBlockEncoding);
219    } else {
220      this.indexBlockEncoder = NoOpIndexBlockEncoder.INSTANCE;
221    }
222    closeOutputStream = path != null;
223    this.cacheConf = cacheConf;
224    this.conf = conf;
225    float encodeBlockSizeRatio = conf.getFloat(UNIFIED_ENCODED_BLOCKSIZE_RATIO, 0f);
226    this.encodedBlockSizeLimit = (int) (hFileContext.getBlocksize() * encodeBlockSizeRatio);
227
228    finishInit(conf);
229    if (LOG.isTraceEnabled()) {
230      LOG.trace("Writer" + (path != null ? " for " + path : "") + " initialized with cacheConf: "
231        + cacheConf + " fileContext: " + fileContext);
232    }
233  }
234
235  /**
236   * Add to the file info. All added key/value pairs can be obtained using
237   * {@link HFile.Reader#getHFileInfo()}.
238   * @param k Key
239   * @param v Value
240   * @throws IOException in case the key or the value are invalid
241   */
242  @Override
243  public void appendFileInfo(final byte[] k, final byte[] v) throws IOException {
244    fileInfo.append(k, v, true);
245  }
246
247  /**
248   * Sets the file info offset in the trailer, finishes up populating fields in the file info, and
249   * writes the file info into the given data output. The reason the data output is not always
250   * {@link #outputStream} is that we store file info as a block in version 2.
251   * @param trailer fixed file trailer
252   * @param out     the data output to write the file info to
253   */
254  protected final void writeFileInfo(FixedFileTrailer trailer, DataOutputStream out)
255    throws IOException {
256    trailer.setFileInfoOffset(outputStream.getPos());
257    finishFileInfo();
258    long startTime = EnvironmentEdgeManager.currentTime();
259    fileInfo.write(out);
260    HFile.updateWriteLatency(EnvironmentEdgeManager.currentTime() - startTime);
261  }
262
263  public long getPos() throws IOException {
264    return outputStream.getPos();
265
266  }
267
268  /**
269   * Checks that the given Cell's key does not violate the key order.
270   * @param cell Cell whose key to check.
271   * @return true if the key is duplicate
272   * @throws IOException if the key or the key order is wrong
273   */
274  protected boolean checkKey(final Cell cell) throws IOException {
275    boolean isDuplicateKey = false;
276
277    if (cell == null) {
278      throw new IOException("Key cannot be null or empty");
279    }
280    if (lastCell != null) {
281      int keyComp = PrivateCellUtil.compareKeyIgnoresMvcc(this.hFileContext.getCellComparator(),
282        lastCell, cell);
283      if (keyComp > 0) {
284        String message = getLexicalErrorMessage(cell);
285        throw new IOException(message);
286      } else if (keyComp == 0) {
287        isDuplicateKey = true;
288      }
289    }
290    return isDuplicateKey;
291  }
292
293  private String getLexicalErrorMessage(Cell cell) {
294    StringBuilder sb = new StringBuilder();
295    sb.append("Added a key not lexically larger than previous. Current cell = ");
296    sb.append(cell);
297    sb.append(", lastCell = ");
298    sb.append(lastCell);
299    // file context includes HFile path and optionally table and CF of file being written
300    sb.append("fileContext=");
301    sb.append(hFileContext);
302    return sb.toString();
303  }
304
305  /** Checks the given value for validity. */
306  protected void checkValue(final byte[] value, final int offset, final int length)
307    throws IOException {
308    if (value == null) {
309      throw new IOException("Value cannot be null");
310    }
311  }
312
313  /** Returns Path or null if we were passed a stream rather than a Path. */
314  @Override
315  public Path getPath() {
316    return path;
317  }
318
319  @Override
320  public String toString() {
321    return "writer=" + (path != null ? path.toString() : null) + ", name=" + name + ", compression="
322      + hFileContext.getCompression().getName();
323  }
324
325  public static Compression.Algorithm compressionByName(String algoName) {
326    if (algoName == null) {
327      return HFile.DEFAULT_COMPRESSION_ALGORITHM;
328    }
329    return Compression.getCompressionAlgorithmByName(algoName);
330  }
331
332  /** A helper method to create HFile output streams in constructors */
333  protected static FSDataOutputStream createOutputStream(Configuration conf, FileSystem fs,
334    Path path, InetSocketAddress[] favoredNodes) throws IOException {
335    FsPermission perms = CommonFSUtils.getFilePermissions(fs, conf, HConstants.DATA_FILE_UMASK_KEY);
336    return FSUtils.create(conf, fs, path, perms, favoredNodes);
337  }
338
339  /** Additional initialization steps */
340  protected void finishInit(final Configuration conf) {
341    if (blockWriter != null) {
342      throw new IllegalStateException("finishInit called twice");
343    }
344    blockWriter =
345      new HFileBlock.Writer(conf, blockEncoder, hFileContext, cacheConf.getByteBuffAllocator(),
346        conf.getInt(MAX_BLOCK_SIZE_UNCOMPRESSED, hFileContext.getBlocksize() * 10));
347    // Data block index writer
348    boolean cacheIndexesOnWrite = cacheConf.shouldCacheIndexesOnWrite();
349    dataBlockIndexWriter = new HFileBlockIndex.BlockIndexWriter(blockWriter,
350      cacheIndexesOnWrite ? cacheConf : null, cacheIndexesOnWrite ? name : null, indexBlockEncoder);
351    dataBlockIndexWriter.setMaxChunkSize(HFileBlockIndex.getMaxChunkSize(conf));
352    dataBlockIndexWriter.setMinIndexNumEntries(HFileBlockIndex.getMinIndexNumEntries(conf));
353    inlineBlockWriters.add(dataBlockIndexWriter);
354
355    // Meta data block index writer
356    metaBlockIndexWriter = new HFileBlockIndex.BlockIndexWriter();
357    LOG.trace("Initialized with {}", cacheConf);
358  }
359
360  /**
361   * At a block boundary, write all the inline blocks and opens new block.
362   */
363  protected void checkBlockBoundary() throws IOException {
364    boolean shouldFinishBlock = false;
365    // This means hbase.writer.unified.encoded.blocksize.ratio was set to something different from 0
366    // and we should use the encoding ratio
367    if (encodedBlockSizeLimit > 0) {
368      shouldFinishBlock = blockWriter.encodedBlockSizeWritten() >= encodedBlockSizeLimit;
369    } else {
370      shouldFinishBlock = blockWriter.encodedBlockSizeWritten() >= hFileContext.getBlocksize()
371        || blockWriter.blockSizeWritten() >= hFileContext.getBlocksize();
372    }
373    shouldFinishBlock &= blockWriter.checkBoundariesWithPredicate();
374    if (shouldFinishBlock) {
375      finishBlock();
376      writeInlineBlocks(false);
377      newBlock();
378    }
379  }
380
381  /** Clean up the data block that is currently being written. */
382  private void finishBlock() throws IOException {
383    if (!blockWriter.isWriting() || blockWriter.blockSizeWritten() == 0) {
384      return;
385    }
386
387    // Update the first data block offset if UNSET; used scanning.
388    if (firstDataBlockOffset == UNSET) {
389      firstDataBlockOffset = outputStream.getPos();
390    }
391    // Update the last data block offset each time through here.
392    lastDataBlockOffset = outputStream.getPos();
393    blockWriter.writeHeaderAndData(outputStream);
394    int onDiskSize = blockWriter.getOnDiskSizeWithHeader();
395    ExtendedCell indexEntry =
396      getMidpoint(this.hFileContext.getCellComparator(), lastCellOfPreviousBlock, firstCellInBlock);
397    dataBlockIndexWriter.addEntry(PrivateCellUtil.getCellKeySerializedAsKeyValueKey(indexEntry),
398      lastDataBlockOffset, onDiskSize);
399    totalUncompressedBytes += blockWriter.getUncompressedSizeWithHeader();
400    if (cacheConf.shouldCacheDataOnWrite()) {
401      doCacheOnWrite(lastDataBlockOffset);
402    }
403  }
404
405  /**
406   * Try to return a Cell that falls between <code>left</code> and <code>right</code> but that is
407   * shorter; i.e. takes up less space. This trick is used building HFile block index. Its an
408   * optimization. It does not always work. In this case we'll just return the <code>right</code>
409   * cell.
410   * @return A cell that sorts between <code>left</code> and <code>right</code>.
411   */
412  public static ExtendedCell getMidpoint(final CellComparator comparator, final ExtendedCell left,
413    final ExtendedCell right) {
414    if (right == null) {
415      throw new IllegalArgumentException("right cell can not be null");
416    }
417    if (left == null) {
418      return right;
419    }
420    // If Cells from meta table, don't mess around. meta table Cells have schema
421    // (table,startrow,hash) so can't be treated as plain byte arrays. Just skip
422    // out without trying to do this optimization.
423    if (comparator instanceof MetaCellComparator) {
424      return right;
425    }
426    byte[] midRow;
427    boolean bufferBacked =
428      left instanceof ByteBufferExtendedCell && right instanceof ByteBufferExtendedCell;
429    if (bufferBacked) {
430      midRow = getMinimumMidpointArray(((ByteBufferExtendedCell) left).getRowByteBuffer(),
431        ((ByteBufferExtendedCell) left).getRowPosition(), left.getRowLength(),
432        ((ByteBufferExtendedCell) right).getRowByteBuffer(),
433        ((ByteBufferExtendedCell) right).getRowPosition(), right.getRowLength());
434    } else {
435      midRow = getMinimumMidpointArray(left.getRowArray(), left.getRowOffset(), left.getRowLength(),
436        right.getRowArray(), right.getRowOffset(), right.getRowLength());
437    }
438    if (midRow != null) {
439      return PrivateCellUtil.createFirstOnRow(midRow);
440    }
441    // Rows are same. Compare on families.
442    if (bufferBacked) {
443      midRow = getMinimumMidpointArray(((ByteBufferExtendedCell) left).getFamilyByteBuffer(),
444        ((ByteBufferExtendedCell) left).getFamilyPosition(), left.getFamilyLength(),
445        ((ByteBufferExtendedCell) right).getFamilyByteBuffer(),
446        ((ByteBufferExtendedCell) right).getFamilyPosition(), right.getFamilyLength());
447    } else {
448      midRow = getMinimumMidpointArray(left.getFamilyArray(), left.getFamilyOffset(),
449        left.getFamilyLength(), right.getFamilyArray(), right.getFamilyOffset(),
450        right.getFamilyLength());
451    }
452    if (midRow != null) {
453      return PrivateCellUtil.createFirstOnRowFamily(right, midRow, 0, midRow.length);
454    }
455    // Families are same. Compare on qualifiers.
456    if (bufferBacked) {
457      midRow = getMinimumMidpointArray(((ByteBufferExtendedCell) left).getQualifierByteBuffer(),
458        ((ByteBufferExtendedCell) left).getQualifierPosition(), left.getQualifierLength(),
459        ((ByteBufferExtendedCell) right).getQualifierByteBuffer(),
460        ((ByteBufferExtendedCell) right).getQualifierPosition(), right.getQualifierLength());
461    } else {
462      midRow = getMinimumMidpointArray(left.getQualifierArray(), left.getQualifierOffset(),
463        left.getQualifierLength(), right.getQualifierArray(), right.getQualifierOffset(),
464        right.getQualifierLength());
465    }
466    if (midRow != null) {
467      return PrivateCellUtil.createFirstOnRowCol(right, midRow, 0, midRow.length);
468    }
469    // No opportunity for optimization. Just return right key.
470    return right;
471  }
472
473  /**
474   * Try to get a byte array that falls between left and right as short as possible with
475   * lexicographical order;
476   * @return Return a new array that is between left and right and minimally sized else just return
477   *         null if left == right.
478   */
479  private static byte[] getMinimumMidpointArray(final byte[] leftArray, final int leftOffset,
480    final int leftLength, final byte[] rightArray, final int rightOffset, final int rightLength) {
481    int minLength = leftLength < rightLength ? leftLength : rightLength;
482    int diffIdx = 0;
483    for (; diffIdx < minLength; diffIdx++) {
484      byte leftByte = leftArray[leftOffset + diffIdx];
485      byte rightByte = rightArray[rightOffset + diffIdx];
486      if ((leftByte & 0xff) > (rightByte & 0xff)) {
487        throw new IllegalArgumentException("Left byte array sorts after right row; left="
488          + Bytes.toStringBinary(leftArray, leftOffset, leftLength) + ", right="
489          + Bytes.toStringBinary(rightArray, rightOffset, rightLength));
490      } else if (leftByte != rightByte) {
491        break;
492      }
493    }
494    if (diffIdx == minLength) {
495      if (leftLength > rightLength) {
496        // right is prefix of left
497        throw new IllegalArgumentException("Left byte array sorts after right row; left="
498          + Bytes.toStringBinary(leftArray, leftOffset, leftLength) + ", right="
499          + Bytes.toStringBinary(rightArray, rightOffset, rightLength));
500      } else if (leftLength < rightLength) {
501        // left is prefix of right.
502        byte[] minimumMidpointArray = new byte[minLength + 1];
503        System.arraycopy(rightArray, rightOffset, minimumMidpointArray, 0, minLength + 1);
504        minimumMidpointArray[minLength] = 0x00;
505        return minimumMidpointArray;
506      } else {
507        // left == right
508        return null;
509      }
510    }
511    // Note that left[diffIdx] can never be equal to 0xff since left < right
512    byte[] minimumMidpointArray = new byte[diffIdx + 1];
513    System.arraycopy(leftArray, leftOffset, minimumMidpointArray, 0, diffIdx + 1);
514    minimumMidpointArray[diffIdx] = (byte) (minimumMidpointArray[diffIdx] + 1);
515    return minimumMidpointArray;
516  }
517
518  /**
519   * Try to create a new byte array that falls between left and right as short as possible with
520   * lexicographical order.
521   * @return Return a new array that is between left and right and minimally sized else just return
522   *         null if left == right.
523   */
524  private static byte[] getMinimumMidpointArray(ByteBuffer left, int leftOffset, int leftLength,
525    ByteBuffer right, int rightOffset, int rightLength) {
526    int minLength = leftLength < rightLength ? leftLength : rightLength;
527    int diffIdx = 0;
528    for (; diffIdx < minLength; diffIdx++) {
529      int leftByte = ByteBufferUtils.toByte(left, leftOffset + diffIdx);
530      int rightByte = ByteBufferUtils.toByte(right, rightOffset + diffIdx);
531      if ((leftByte & 0xff) > (rightByte & 0xff)) {
532        throw new IllegalArgumentException("Left byte array sorts after right row; left="
533          + ByteBufferUtils.toStringBinary(left, leftOffset, leftLength) + ", right="
534          + ByteBufferUtils.toStringBinary(right, rightOffset, rightLength));
535      } else if (leftByte != rightByte) {
536        break;
537      }
538    }
539    if (diffIdx == minLength) {
540      if (leftLength > rightLength) {
541        // right is prefix of left
542        throw new IllegalArgumentException("Left byte array sorts after right row; left="
543          + ByteBufferUtils.toStringBinary(left, leftOffset, leftLength) + ", right="
544          + ByteBufferUtils.toStringBinary(right, rightOffset, rightLength));
545      } else if (leftLength < rightLength) {
546        // left is prefix of right.
547        byte[] minimumMidpointArray = new byte[minLength + 1];
548        ByteBufferUtils.copyFromBufferToArray(minimumMidpointArray, right, rightOffset, 0,
549          minLength + 1);
550        minimumMidpointArray[minLength] = 0x00;
551        return minimumMidpointArray;
552      } else {
553        // left == right
554        return null;
555      }
556    }
557    // Note that left[diffIdx] can never be equal to 0xff since left < right
558    byte[] minimumMidpointArray = new byte[diffIdx + 1];
559    ByteBufferUtils.copyFromBufferToArray(minimumMidpointArray, left, leftOffset, 0, diffIdx + 1);
560    minimumMidpointArray[diffIdx] = (byte) (minimumMidpointArray[diffIdx] + 1);
561    return minimumMidpointArray;
562  }
563
564  /** Gives inline block writers an opportunity to contribute blocks. */
565  private void writeInlineBlocks(boolean closing) throws IOException {
566    for (InlineBlockWriter ibw : inlineBlockWriters) {
567      while (ibw.shouldWriteBlock(closing)) {
568        long offset = outputStream.getPos();
569        boolean cacheThisBlock = ibw.getCacheOnWrite();
570        ibw.writeInlineBlock(blockWriter.startWriting(ibw.getInlineBlockType()));
571        blockWriter.writeHeaderAndData(outputStream);
572        ibw.blockWritten(offset, blockWriter.getOnDiskSizeWithHeader(),
573          blockWriter.getUncompressedSizeWithoutHeader());
574        totalUncompressedBytes += blockWriter.getUncompressedSizeWithHeader();
575
576        if (cacheThisBlock) {
577          doCacheOnWrite(offset);
578        }
579      }
580    }
581  }
582
583  /**
584   * Caches the last written HFile block.
585   * @param offset the offset of the block we want to cache. Used to determine the cache key.
586   */
587  private void doCacheOnWrite(long offset) {
588    cacheConf.getBlockCache().ifPresent(cache -> {
589      HFileBlock cacheFormatBlock = blockWriter.getBlockForCaching(cacheConf);
590      try {
591        BlockCacheKey key = buildCacheBlockKey(offset, cacheFormatBlock.getBlockType());
592        if (!shouldCacheBlock(cache, key)) {
593          return;
594        }
595        cache.cacheBlock(key, cacheFormatBlock, cacheConf.isInMemory(), true);
596      } finally {
597        // refCnt will auto increase when block add to Cache, see RAMCache#putIfAbsent
598        cacheFormatBlock.release();
599      }
600    });
601  }
602
603  private BlockCacheKey buildCacheBlockKey(long offset, BlockType blockType) {
604    if (path != null) {
605      return new BlockCacheKey(name, familyName, regionName, offset, true, blockType, false);
606    }
607    return new BlockCacheKey(name, offset, true, blockType);
608  }
609
610  private boolean shouldCacheBlock(BlockCache cache, BlockCacheKey key) {
611    Optional<Boolean> result =
612      cache.shouldCacheBlock(key, timeRangeTrackerForTiering.get().getMax(), conf);
613    return result.orElse(true);
614  }
615
616  /**
617   * Ready a new block for writing.
618   */
619  protected void newBlock() throws IOException {
620    // This is where the next block begins.
621    blockWriter.startWriting(BlockType.DATA);
622    firstCellInBlock = null;
623    if (lastCell != null) {
624      lastCellOfPreviousBlock = lastCell;
625    }
626  }
627
628  /**
629   * Add a meta block to the end of the file. Call before close(). Metadata blocks are expensive.
630   * Fill one with a bunch of serialized data rather than do a metadata block per metadata instance.
631   * If metadata is small, consider adding to file info using
632   * {@link #appendFileInfo(byte[], byte[])} name of the block will call readFields to get data
633   * later (DO NOT REUSE)
634   */
635  @Override
636  public void appendMetaBlock(String metaBlockName, Writable content) {
637    byte[] key = Bytes.toBytes(metaBlockName);
638    int i;
639    for (i = 0; i < metaNames.size(); ++i) {
640      // stop when the current key is greater than our own
641      byte[] cur = metaNames.get(i);
642      if (Bytes.BYTES_RAWCOMPARATOR.compare(cur, 0, cur.length, key, 0, key.length) > 0) {
643        break;
644      }
645    }
646    metaNames.add(i, key);
647    metaData.add(i, content);
648  }
649
650  @Override
651  public void close() throws IOException {
652    if (outputStream == null) {
653      return;
654    }
655    // Save data block encoder metadata in the file info.
656    blockEncoder.saveMetadata(this);
657    // Save index block encoder metadata in the file info.
658    indexBlockEncoder.saveMetadata(this);
659    // Write out the end of the data blocks, then write meta data blocks.
660    // followed by fileinfo, data block index and meta block index.
661
662    finishBlock();
663    writeInlineBlocks(true);
664
665    FixedFileTrailer trailer = new FixedFileTrailer(getMajorVersion(), getMinorVersion());
666
667    // Write out the metadata blocks if any.
668    if (!metaNames.isEmpty()) {
669      for (int i = 0; i < metaNames.size(); ++i) {
670        // store the beginning offset
671        long offset = outputStream.getPos();
672        // write the metadata content
673        DataOutputStream dos = blockWriter.startWriting(BlockType.META);
674        metaData.get(i).write(dos);
675
676        blockWriter.writeHeaderAndData(outputStream);
677        totalUncompressedBytes += blockWriter.getUncompressedSizeWithHeader();
678
679        // Add the new meta block to the meta index.
680        metaBlockIndexWriter.addEntry(metaNames.get(i), offset,
681          blockWriter.getOnDiskSizeWithHeader());
682      }
683    }
684
685    // Load-on-open section.
686
687    // Data block index.
688    //
689    // In version 2, this section of the file starts with the root level data
690    // block index. We call a function that writes intermediate-level blocks
691    // first, then root level, and returns the offset of the root level block
692    // index.
693
694    long rootIndexOffset = dataBlockIndexWriter.writeIndexBlocks(outputStream);
695    trailer.setLoadOnOpenOffset(rootIndexOffset);
696
697    // Meta block index.
698    metaBlockIndexWriter.writeSingleLevelIndex(blockWriter.startWriting(BlockType.ROOT_INDEX),
699      "meta");
700    blockWriter.writeHeaderAndData(outputStream);
701    totalUncompressedBytes += blockWriter.getUncompressedSizeWithHeader();
702
703    if (this.hFileContext.isIncludesMvcc()) {
704      appendFileInfo(MAX_MEMSTORE_TS_KEY, Bytes.toBytes(maxMemstoreTS));
705      appendFileInfo(KEY_VALUE_VERSION, Bytes.toBytes(KEY_VALUE_VER_WITH_MEMSTORE));
706    }
707
708    // File info
709    writeFileInfo(trailer, blockWriter.startWriting(BlockType.FILE_INFO));
710    blockWriter.writeHeaderAndData(outputStream);
711    totalUncompressedBytes += blockWriter.getUncompressedSizeWithHeader();
712
713    // Load-on-open data supplied by higher levels, e.g. Bloom filters.
714    for (BlockWritable w : additionalLoadOnOpenData) {
715      blockWriter.writeBlock(w, outputStream);
716      totalUncompressedBytes += blockWriter.getUncompressedSizeWithHeader();
717    }
718
719    // Now finish off the trailer.
720    trailer.setNumDataIndexLevels(dataBlockIndexWriter.getNumLevels());
721    trailer.setUncompressedDataIndexSize(dataBlockIndexWriter.getTotalUncompressedSize());
722    trailer.setFirstDataBlockOffset(firstDataBlockOffset);
723    trailer.setLastDataBlockOffset(lastDataBlockOffset);
724    trailer.setComparatorClass(this.hFileContext.getCellComparator().getClass());
725    trailer.setDataIndexCount(dataBlockIndexWriter.getNumRootEntries());
726
727    finishClose(trailer);
728
729    blockWriter.release();
730  }
731
732  @Override
733  public void addInlineBlockWriter(InlineBlockWriter ibw) {
734    inlineBlockWriters.add(ibw);
735  }
736
737  @Override
738  public void addGeneralBloomFilter(final BloomFilterWriter bfw) {
739    this.addBloomFilter(bfw, BlockType.GENERAL_BLOOM_META);
740  }
741
742  @Override
743  public void addDeleteFamilyBloomFilter(final BloomFilterWriter bfw) {
744    this.addBloomFilter(bfw, BlockType.DELETE_FAMILY_BLOOM_META);
745  }
746
747  private void addBloomFilter(final BloomFilterWriter bfw, final BlockType blockType) {
748    if (bfw.getKeyCount() <= 0) {
749      return;
750    }
751
752    if (
753      blockType != BlockType.GENERAL_BLOOM_META && blockType != BlockType.DELETE_FAMILY_BLOOM_META
754    ) {
755      throw new RuntimeException("Block Type: " + blockType.toString() + "is not supported");
756    }
757    additionalLoadOnOpenData.add(new BlockWritable() {
758      @Override
759      public BlockType getBlockType() {
760        return blockType;
761      }
762
763      @Override
764      public void writeToBlock(DataOutput out) throws IOException {
765        bfw.getMetaWriter().write(out);
766        Writable dataWriter = bfw.getDataWriter();
767        if (dataWriter != null) {
768          dataWriter.write(out);
769        }
770      }
771    });
772  }
773
774  @Override
775  public HFileContext getFileContext() {
776    return hFileContext;
777  }
778
779  /**
780   * Add key/value to file. Keys must be added in an order that agrees with the Comparator passed on
781   * construction. Cell to add. Cannot be empty nor null.
782   */
783  @Override
784  public void append(final ExtendedCell cell) throws IOException {
785    // checkKey uses comparator to check we are writing in order.
786    boolean dupKey = checkKey(cell);
787    if (!dupKey) {
788      checkBlockBoundary();
789    }
790
791    if (!blockWriter.isWriting()) {
792      newBlock();
793    }
794
795    blockWriter.write(cell);
796
797    totalKeyLength += PrivateCellUtil.estimatedSerializedSizeOfKey(cell);
798    totalValueLength += cell.getValueLength();
799    if (lenOfBiggestCell < PrivateCellUtil.estimatedSerializedSizeOf(cell)) {
800      lenOfBiggestCell = PrivateCellUtil.estimatedSerializedSizeOf(cell);
801      keyOfBiggestCell = PrivateCellUtil.getCellKeySerializedAsKeyValueKey(cell);
802    }
803    // Are we the first key in this block?
804    if (firstCellInBlock == null) {
805      // If cell is big, block will be closed and this firstCellInBlock reference will only last
806      // a short while.
807      firstCellInBlock = cell;
808    }
809
810    // TODO: What if cell is 10MB and we write infrequently? We hold on to cell here indefinitely?
811    lastCell = cell;
812    entryCount++;
813    this.maxMemstoreTS = Math.max(this.maxMemstoreTS, cell.getSequenceId());
814    int tagsLength = cell.getTagsLength();
815    if (tagsLength > this.maxTagsLength) {
816      this.maxTagsLength = tagsLength;
817    }
818
819    trackTimestamps(cell);
820  }
821
822  @Override
823  public void beforeShipped() throws IOException {
824    this.blockWriter.beforeShipped();
825    // Add clone methods for every cell
826    if (this.lastCell != null) {
827      this.lastCell = KeyValueUtil.toNewKeyCell(this.lastCell);
828    }
829    if (this.firstCellInBlock != null) {
830      this.firstCellInBlock = KeyValueUtil.toNewKeyCell(this.firstCellInBlock);
831    }
832    if (this.lastCellOfPreviousBlock != null) {
833      this.lastCellOfPreviousBlock = KeyValueUtil.toNewKeyCell(this.lastCellOfPreviousBlock);
834    }
835  }
836
837  public Cell getLastCell() {
838    return lastCell;
839  }
840
841  protected void finishFileInfo() throws IOException {
842    if (lastCell != null) {
843      // Make a copy. The copy is stuffed into our fileinfo map. Needs a clean
844      // byte buffer. Won't take a tuple.
845      byte[] lastKey = PrivateCellUtil.getCellKeySerializedAsKeyValueKey(this.lastCell);
846      fileInfo.append(HFileInfo.LASTKEY, lastKey, false);
847    }
848
849    // Average key length.
850    int avgKeyLen = entryCount == 0 ? 0 : (int) (totalKeyLength / entryCount);
851    fileInfo.append(HFileInfo.AVG_KEY_LEN, Bytes.toBytes(avgKeyLen), false);
852    fileInfo.append(HFileInfo.CREATE_TIME_TS, Bytes.toBytes(hFileContext.getFileCreateTime()),
853      false);
854
855    // Average value length.
856    int avgValueLen = entryCount == 0 ? 0 : (int) (totalValueLength / entryCount);
857    fileInfo.append(HFileInfo.AVG_VALUE_LEN, Bytes.toBytes(avgValueLen), false);
858
859    // Biggest cell.
860    if (keyOfBiggestCell != null) {
861      fileInfo.append(HFileInfo.KEY_OF_BIGGEST_CELL, keyOfBiggestCell, false);
862      fileInfo.append(HFileInfo.LEN_OF_BIGGEST_CELL, Bytes.toBytes(lenOfBiggestCell), false);
863      LOG.debug("Len of the biggest cell in {} is {}, key is {}",
864        this.getPath() == null ? "" : this.getPath().toString(), lenOfBiggestCell,
865        CellUtil.toString(new KeyValue.KeyOnlyKeyValue(keyOfBiggestCell), false));
866    }
867
868    if (hFileContext.isIncludesTags()) {
869      // When tags are not being written in this file, MAX_TAGS_LEN is excluded
870      // from the FileInfo
871      fileInfo.append(HFileInfo.MAX_TAGS_LEN, Bytes.toBytes(this.maxTagsLength), false);
872      boolean tagsCompressed = (hFileContext.getDataBlockEncoding() != DataBlockEncoding.NONE)
873        && hFileContext.isCompressTags();
874      fileInfo.append(HFileInfo.TAGS_COMPRESSED, Bytes.toBytes(tagsCompressed), false);
875    }
876  }
877
878  protected int getMajorVersion() {
879    return 3;
880  }
881
882  protected int getMinorVersion() {
883    return HFileReaderImpl.MAX_MINOR_VERSION;
884  }
885
886  protected void finishClose(FixedFileTrailer trailer) throws IOException {
887    // Write out encryption metadata before finalizing if we have a valid crypto context
888    Encryption.Context cryptoContext = hFileContext.getEncryptionContext();
889    if (cryptoContext != Encryption.Context.NONE) {
890      // Wrap the context's key and write it as the encryption metadata, the wrapper includes
891      // all information needed for decryption
892      trailer.setEncryptionKey(EncryptionUtil.wrapKey(
893        cryptoContext.getConf(), cryptoContext.getConf()
894          .get(HConstants.CRYPTO_MASTERKEY_NAME_CONF_KEY, User.getCurrent().getShortName()),
895        cryptoContext.getKey()));
896    }
897    // Now we can finish the close
898    trailer.setMetaIndexCount(metaNames.size());
899    trailer.setTotalUncompressedBytes(totalUncompressedBytes + trailer.getTrailerSize());
900    trailer.setEntryCount(entryCount);
901    trailer.setCompressionCodec(hFileContext.getCompression());
902
903    long startTime = EnvironmentEdgeManager.currentTime();
904    trailer.serialize(outputStream);
905    HFile.updateWriteLatency(EnvironmentEdgeManager.currentTime() - startTime);
906
907    if (closeOutputStream) {
908      outputStream.close();
909      outputStream = null;
910    }
911  }
912
913  /**
914   * Add TimestampRange and earliest put timestamp to Metadata
915   */
916  public void appendTrackedTimestampsToMetadata() throws IOException {
917    // TODO: The StoreFileReader always converts the byte[] to TimeRange
918    // via TimeRangeTracker, so we should write the serialization data of TimeRange directly.
919    appendFileInfo(TIMERANGE_KEY, TimeRangeTracker.toByteArray(timeRangeTracker));
920    appendFileInfo(EARLIEST_PUT_TS, Bytes.toBytes(earliestPutTs));
921  }
922
923  public void appendCustomCellTimestampsToMetadata(TimeRangeTracker timeRangeTracker)
924    throws IOException {
925    // TODO: The StoreFileReader always converts the byte[] to TimeRange
926    // via TimeRangeTracker, so we should write the serialization data of TimeRange directly.
927    appendFileInfo(CUSTOM_TIERING_TIME_RANGE, TimeRangeTracker.toByteArray(timeRangeTracker));
928  }
929
930  /**
931   * Record the earliest Put timestamp. If the timeRangeTracker is not set, update TimeRangeTracker
932   * to include the timestamp of this key
933   */
934  private void trackTimestamps(final ExtendedCell cell) {
935    if (KeyValue.Type.Put == KeyValue.Type.codeToType(cell.getTypeByte())) {
936      earliestPutTs = Math.min(earliestPutTs, cell.getTimestamp());
937    }
938    timeRangeTracker.includeTimestamp(cell);
939  }
940}