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}