Summary: This change builds on https://github.com/facebook/rocksdb/issues/13540 and https://github.com/facebook/rocksdb/issues/13626 in allowing a CompressionManager / Compressor / Decompressor to use a custom compression algorithm, with a distinct CompressionType. For background, review the API comments on CompressionManager and its CompatibilityName() function. Highlights: * Reserve and name 127 new CompressionTypes that can be used for custom compression algorithms / schemas. In many or most cases I expect the enumerators such as `kCustomCompression8F` to be used in user code rather than casting between integers and CompressionTypes, as I expect the supported custom compression algorithms to be identifiable / enumerable at compile time. * When using these custom compression types, a CompressionManager must use a CompatibilityName() other than the built-in one AND new format_version=7 (see below). * When building new SST files, track the full set of CompressionTypes actually used (usually just one aside from kNoCompression), using our efficient bitset SmallEnumSet, which supports fast iteration over the bits set to 1. Ideally, to support mixed or non-mixed compression algorithms in a file as efficiently as possible, we would know the set of CompressionTypes as SST file open time. * New schema for `TableProperties::compression_name` in format_version=7 to represent the CompressionManager's CompatibilityName(), the set of CompressionTypes used, and potentially more in the future, while keeping the data relatively human-readable. * It would be possible to do this without a new format_version, but then the only way to ensure incompatible versions fail is with an unsupported CompressionType tag, not with a compression_name property. Therefore, (a) I prefer not to put something misleading in the `compression_name` property (a built-in compression name) when there is nuance because of a CompressionManager, and (b) I prefer better, more consistent error messages that refer to either format_version or the CompressionManager's CompatibilityName(), rather than an unrecognized custom CompressionType value (which could have come from various CompressionManagers). * The current configured CompressionManager is passed in to TableReaders so that it (or one it knows about) can be used if it matches the CompatibilityName() used for compression in the SST file. Until the connection with ObjectRegistry is implemented, the only way to read files generated with a particular CompressionManager using custom compression algorithms is to configure it (or a known relative; see FindCompatibleCompressionManager()) in the ColumnFamilyOptions. * Optimized snappy compression with BuiltinDecompressorV2SnappyOnly, to offset some small added overheads with the new tracking. This is essentially an early part of the planned refactoring that will get rid of the old internal compression APIs. * Another small optimization in eliminating an unnecessary key copy in flush (builder.cc). * Fix some handling of named CompressionManagers in CompressionManager::CreateFromString() (problem seen in https://github.com/facebook/rocksdb/issues/13647) Smaller things: * Adds Name() and GetId() functions to Compressor for debugging/logging purposes. (Compressor and Decompressor are not expected to be Customizable because they are only instantiated by a CompressionManager.) * When using an explicit compression_manager, the GetId() of the CompressionManager and the Compressor used to build the file are stored as bonus entries in the compression_options table property. This table property is not parsed anywhere, so it is currently for human reading, but still could be parsed with the new underscore-prefixed bonus entries. IMHO, this is preferable to additional table properties, which would increase memory fragmentation in the TableProperties objects and likely take slightly more CPU on SST open and slightly more storage. * ReleaseWorkingArea() function from protected to public to make wrappers work, because of a quirk in C++ (vs. Java) in which you cannot access protected members of another instance of the same class (sigh) * Added `CompressionManager:: SupportsCompressionType()` for early options sanity checking. Follow-up before release: * Make format_version=7 official / supported * Stress test coverage Sooner than later: * Update tests for RoundRobinManager and SimpleMixedCompressionManager to take advantage of e.g. set of compression types in compression_name property * ObjectRegistry stuff * Refactor away old internal compression APIs Pull Request resolved: https://github.com/facebook/rocksdb/pull/13659 Test Plan: Basic unit test added. ## Performance ### SST write performance ``` SUFFIX=`tty | sed 's|/|_|g'`; for ARGS in "-compression_type=none" "-compression_type=snappy" "-compression_type=zstd" "-compression_type=snappy -verify_compression=1" "-compression_type=zstd -verify_compression=1" "-compression_type=zstd -compression_max_dict_bytes=8180"; do echo $ARGS; (for I in `seq 1 20`; do BIN=/dev/shm/dbbench${SUFFIX}.bin; rm -f $BIN; cp db_bench $BIN; $BIN -db=/dev/shm/dbbench$SUFFIX --benchmarks=fillseq -num=10000000 -compaction_style=2 -fifo_compaction_max_table_files_size_mb=1000 -fifo_compaction_allow_compaction=0 -disable_wal -write_buffer_size=12000000 -format_version=7 $ARGS 2>&1 | grep micros/op; done) | awk '{n++; sum += $5;} END { print int(sum / n); }'; done ``` Ops/sec, Before -> After, both fv=6: -compression_type=none 1894386 -> 1858403 (-2.0%) -compression_type=snappy 1859131 -> 1807469 (-2.8%) -compression_type=zstd 1191428 -> 1214374 (+1.9%) -compression_type=snappy -verify_compression=1 1861819 -> 1858342 (+0.2%) -compression_type=zstd -verify_compression=1 979435 -> 995870 (+1.6%) -compression_type=zstd -compression_max_dict_bytes=8180 905349 -> 940563 (+3.9%) Ops/sec, Before fv=6 -> After fv=7: -compression_type=none 1879365 -> 1836159 (-2.3%) -compression_type=snappy 1865460 -> 1830916 (-1.9%) -compression_type=zstd 1191428 -> 1210260 (+1.6%) -compression_type=snappy -verify_compression=1 1866756 -> 1818989 (-2.6%) -compression_type=zstd -verify_compression=1 982640 -> 997129 (+1.5%) -compression_type=zstd -compression_max_dict_bytes=8180 912608 -> 937248 (+2.7%) ### SST read performance Create DBs ``` for COMP in none snappy zstd; do echo $ARGS; ./db_bench -db=/dev/shm/dbbench-7-$COMP --benchmarks=fillseq,flush -num=10000000 -compaction_style=2 -fifo_compaction_max_table_files_size_mb=1000 -fifo_compaction_allow_compaction=0 -disable_wal -write_buffer_size=12000000 -compression_type=$COMP -format_version=7; done ``` And test ``` for COMP in none snappy zstd none; do echo $COMP; (for I in `seq 1 8`; do ./db_bench -readonly -db=/dev/shm/dbbench -7-$COMP --benchmarks=readrandom -num=10000000 -duration=20 -threads=8 2>&1 | grep micros/op; done ) | awk '{n++; sum += $5;} END { print int(sum / n); }'; done ``` Ops/sec, Before -> After (both fv=6) none 1491732 -> 1500209 (+0.6%) snappy 1157216 -> 1169202 (+1.0%) zstd 695414 -> 703719 (+1.2%) none (again) 1491787 -> 1528789 (+2.4%) Ops/sec, Before fv=6 -> After fv=7: none 1492278 -> 1508668 (+1.1%) snappy 1140769 -> 1152613 (+1.0%) zstd 696437 -> 696511 (+0.0%) none (again) 1500585 -> 1512037 (+0.7%) Overall, I think we can take the read CPU improvement in exchange for the hit (in some cases) on background write CPU Reviewed By: hx235 Differential Revision: D76520739 Pulled By: pdillinger fbshipit-source-id: e73bd72502ff85c8779cba313f26f7d1fd50be3a
754 lines
28 KiB
C++
754 lines
28 KiB
C++
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
|
|
// This source code is licensed under both the GPLv2 (found in the
|
|
// COPYING file in the root directory) and Apache 2.0 License
|
|
// (found in the LICENSE.Apache file in the root directory).
|
|
//
|
|
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
|
|
// Use of this source code is governed by a BSD-style license that can be
|
|
// found in the LICENSE file. See the AUTHORS file for names of contributors.
|
|
|
|
#include "table/format.h"
|
|
|
|
#include <cinttypes>
|
|
#include <cstdint>
|
|
#include <string>
|
|
|
|
#include "block_fetcher.h"
|
|
#include "file/random_access_file_reader.h"
|
|
#include "memory/memory_allocator_impl.h"
|
|
#include "monitoring/perf_context_imp.h"
|
|
#include "monitoring/statistics_impl.h"
|
|
#include "options/options_helper.h"
|
|
#include "port/likely.h"
|
|
#include "rocksdb/env.h"
|
|
#include "rocksdb/options.h"
|
|
#include "rocksdb/table.h"
|
|
#include "table/block_based/block.h"
|
|
#include "table/block_based/block_based_table_reader.h"
|
|
#include "table/persistent_cache_helper.h"
|
|
#include "unique_id_impl.h"
|
|
#include "util/cast_util.h"
|
|
#include "util/coding.h"
|
|
#include "util/compression.h"
|
|
#include "util/crc32c.h"
|
|
#include "util/hash.h"
|
|
#include "util/stop_watch.h"
|
|
#include "util/string_util.h"
|
|
#include "util/xxhash.h"
|
|
|
|
namespace ROCKSDB_NAMESPACE {
|
|
|
|
const char* kHostnameForDbHostId = "__hostname__";
|
|
|
|
bool ShouldReportDetailedTime(Env* env, Statistics* stats) {
|
|
return env != nullptr && stats != nullptr &&
|
|
stats->get_stats_level() > kExceptDetailedTimers;
|
|
}
|
|
|
|
void BlockHandle::EncodeTo(std::string* dst) const {
|
|
// Sanity check that all fields have been set
|
|
assert(offset_ != ~uint64_t{0});
|
|
assert(size_ != ~uint64_t{0});
|
|
PutVarint64Varint64(dst, offset_, size_);
|
|
}
|
|
|
|
char* BlockHandle::EncodeTo(char* dst) const {
|
|
// Sanity check that all fields have been set
|
|
assert(offset_ != ~uint64_t{0});
|
|
assert(size_ != ~uint64_t{0});
|
|
char* cur = EncodeVarint64(dst, offset_);
|
|
cur = EncodeVarint64(cur, size_);
|
|
return cur;
|
|
}
|
|
|
|
Status BlockHandle::DecodeFrom(Slice* input) {
|
|
if (GetVarint64(input, &offset_) && GetVarint64(input, &size_)) {
|
|
return Status::OK();
|
|
} else {
|
|
// reset in case failure after partially decoding
|
|
offset_ = 0;
|
|
size_ = 0;
|
|
return Status::Corruption("bad block handle");
|
|
}
|
|
}
|
|
|
|
Status BlockHandle::DecodeSizeFrom(uint64_t _offset, Slice* input) {
|
|
if (GetVarint64(input, &size_)) {
|
|
offset_ = _offset;
|
|
return Status::OK();
|
|
} else {
|
|
// reset in case failure after partially decoding
|
|
offset_ = 0;
|
|
size_ = 0;
|
|
return Status::Corruption("bad block handle");
|
|
}
|
|
}
|
|
|
|
// Return a string that contains the copy of handle.
|
|
std::string BlockHandle::ToString(bool hex) const {
|
|
std::string handle_str;
|
|
EncodeTo(&handle_str);
|
|
if (hex) {
|
|
return Slice(handle_str).ToString(true);
|
|
} else {
|
|
return handle_str;
|
|
}
|
|
}
|
|
|
|
const BlockHandle BlockHandle::kNullBlockHandle(0, 0);
|
|
|
|
void IndexValue::EncodeTo(std::string* dst, bool have_first_key,
|
|
const BlockHandle* previous_handle) const {
|
|
if (previous_handle) {
|
|
// WART: this is specific to Block-based table
|
|
assert(handle.offset() == previous_handle->offset() +
|
|
previous_handle->size() +
|
|
BlockBasedTable::kBlockTrailerSize);
|
|
PutVarsignedint64(dst, handle.size() - previous_handle->size());
|
|
} else {
|
|
handle.EncodeTo(dst);
|
|
}
|
|
assert(dst->size() != 0);
|
|
|
|
if (have_first_key) {
|
|
PutLengthPrefixedSlice(dst, first_internal_key);
|
|
}
|
|
}
|
|
|
|
Status IndexValue::DecodeFrom(Slice* input, bool have_first_key,
|
|
const BlockHandle* previous_handle) {
|
|
if (previous_handle) {
|
|
int64_t delta;
|
|
if (!GetVarsignedint64(input, &delta)) {
|
|
return Status::Corruption("bad delta-encoded index value");
|
|
}
|
|
// WART: this is specific to Block-based table
|
|
handle = BlockHandle(previous_handle->offset() + previous_handle->size() +
|
|
BlockBasedTable::kBlockTrailerSize,
|
|
previous_handle->size() + delta);
|
|
} else {
|
|
Status s = handle.DecodeFrom(input);
|
|
if (!s.ok()) {
|
|
return s;
|
|
}
|
|
}
|
|
|
|
if (!have_first_key) {
|
|
first_internal_key = Slice();
|
|
} else if (!GetLengthPrefixedSlice(input, &first_internal_key)) {
|
|
return Status::Corruption("bad first key in block info");
|
|
}
|
|
|
|
return Status::OK();
|
|
}
|
|
|
|
std::string IndexValue::ToString(bool hex, bool have_first_key) const {
|
|
std::string s;
|
|
EncodeTo(&s, have_first_key, nullptr);
|
|
if (hex) {
|
|
return Slice(s).ToString(true);
|
|
} else {
|
|
return s;
|
|
}
|
|
}
|
|
|
|
namespace {
|
|
inline bool IsLegacyFooterFormat(uint64_t magic_number) {
|
|
return magic_number == kLegacyBlockBasedTableMagicNumber ||
|
|
magic_number == kLegacyPlainTableMagicNumber;
|
|
}
|
|
inline uint64_t UpconvertLegacyFooterFormat(uint64_t magic_number) {
|
|
if (magic_number == kLegacyBlockBasedTableMagicNumber) {
|
|
return kBlockBasedTableMagicNumber;
|
|
}
|
|
if (magic_number == kLegacyPlainTableMagicNumber) {
|
|
return kPlainTableMagicNumber;
|
|
}
|
|
assert(false);
|
|
return magic_number;
|
|
}
|
|
inline uint64_t DownconvertToLegacyFooterFormat(uint64_t magic_number) {
|
|
if (magic_number == kBlockBasedTableMagicNumber) {
|
|
return kLegacyBlockBasedTableMagicNumber;
|
|
}
|
|
if (magic_number == kPlainTableMagicNumber) {
|
|
return kLegacyPlainTableMagicNumber;
|
|
}
|
|
assert(false);
|
|
return magic_number;
|
|
}
|
|
inline uint8_t BlockTrailerSizeForMagicNumber(uint64_t magic_number) {
|
|
if (magic_number == kBlockBasedTableMagicNumber ||
|
|
magic_number == kLegacyBlockBasedTableMagicNumber) {
|
|
return static_cast<uint8_t>(BlockBasedTable::kBlockTrailerSize);
|
|
} else {
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
// Footer format, in three parts:
|
|
// * Part1
|
|
// -> format_version == 0 (inferred from legacy magic number)
|
|
// <empty> (0 bytes)
|
|
// -> format_version >= 1
|
|
// checksum type (char, 1 byte)
|
|
// * Part2
|
|
// -> format_version <= 5
|
|
// metaindex handle (varint64 offset, varint64 size)
|
|
// index handle (varint64 offset, varint64 size)
|
|
// <zero padding> for part2 size = 2 * BlockHandle::kMaxEncodedLength = 40
|
|
// - This padding is unchecked/ignored
|
|
// -> format_version >= 6
|
|
// extended magic number (4 bytes) = 0x3e 0x00 0x7a 0x00
|
|
// - Also surely invalid (size 0) handles if interpreted as older version
|
|
// - (Helps ensure a corrupted format_version doesn't get us far with no
|
|
// footer checksum.)
|
|
// footer_checksum (uint32LE, 4 bytes)
|
|
// - Checksum of above checksum type of whole footer, with this field
|
|
// set to all zeros.
|
|
// base_context_checksum (uint32LE, 4 bytes)
|
|
// metaindex block size (uint32LE, 4 bytes)
|
|
// - Assumed to be immediately before footer, < 4GB
|
|
// <zero padding> (24 bytes, reserved for future use)
|
|
// - Brings part2 size also to 40 bytes
|
|
// - Checked that last eight bytes == 0, so reserved for a future
|
|
// incompatible feature (but under format_version=6)
|
|
// * Part3
|
|
// -> format_version == 0 (inferred from legacy magic number)
|
|
// legacy magic number (8 bytes)
|
|
// -> format_version >= 1 (inferred from NOT legacy magic number)
|
|
// format_version (uint32LE, 4 bytes), also called "footer version"
|
|
// newer magic number (8 bytes)
|
|
const std::array<char, 4> kExtendedMagic{{0x3e, 0x00, 0x7a, 0x00}};
|
|
constexpr size_t kFooterPart2Size = 2 * BlockHandle::kMaxEncodedLength;
|
|
} // namespace
|
|
|
|
Status FooterBuilder::Build(uint64_t magic_number, uint32_t format_version,
|
|
uint64_t footer_offset, ChecksumType checksum_type,
|
|
const BlockHandle& metaindex_handle,
|
|
const BlockHandle& index_handle,
|
|
uint32_t base_context_checksum) {
|
|
assert(magic_number != Footer::kNullTableMagicNumber);
|
|
assert(IsSupportedFormatVersion(format_version) ||
|
|
TEST_AllowUnsupportedFormatVersion());
|
|
|
|
char* part2;
|
|
char* part3;
|
|
if (format_version > 0) {
|
|
slice_ = Slice(data_.data(), Footer::kNewVersionsEncodedLength);
|
|
// Generate parts 1 and 3
|
|
char* cur = data_.data();
|
|
// Part 1
|
|
*(cur++) = checksum_type;
|
|
// Part 2
|
|
part2 = cur;
|
|
// Skip over part 2 for now
|
|
cur += kFooterPart2Size;
|
|
// Part 3
|
|
part3 = cur;
|
|
EncodeFixed32(cur, format_version);
|
|
cur += 4;
|
|
EncodeFixed64(cur, magic_number);
|
|
assert(cur + 8 == slice_.data() + slice_.size());
|
|
} else {
|
|
slice_ = Slice(data_.data(), Footer::kVersion0EncodedLength);
|
|
// Legacy SST files use kCRC32c checksum but it's not stored in footer.
|
|
assert(checksum_type == kNoChecksum || checksum_type == kCRC32c);
|
|
// Generate part 3 (part 1 empty, skip part 2 for now)
|
|
part2 = data_.data();
|
|
part3 = part2 + kFooterPart2Size;
|
|
char* cur = part3;
|
|
// Use legacy magic numbers to indicate format_version=0, for
|
|
// compatibility. No other cases should use format_version=0.
|
|
EncodeFixed64(cur, DownconvertToLegacyFooterFormat(magic_number));
|
|
assert(cur + 8 == slice_.data() + slice_.size());
|
|
}
|
|
|
|
if (format_version >= 6) {
|
|
if (BlockTrailerSizeForMagicNumber(magic_number) != 0) {
|
|
// base context checksum required for table formats with block checksums
|
|
assert(base_context_checksum != 0);
|
|
assert(ChecksumModifierForContext(base_context_checksum, 0) != 0);
|
|
} else {
|
|
// base context checksum not used
|
|
assert(base_context_checksum == 0);
|
|
assert(ChecksumModifierForContext(base_context_checksum, 0) == 0);
|
|
}
|
|
|
|
// Start populating Part 2
|
|
char* cur = data_.data() + /* part 1 size */ 1;
|
|
// Set extended magic of part2
|
|
std::copy(kExtendedMagic.begin(), kExtendedMagic.end(), cur);
|
|
cur += kExtendedMagic.size();
|
|
// Fill checksum data with zeros (for later computing checksum)
|
|
char* checksum_data = cur;
|
|
EncodeFixed32(cur, 0);
|
|
cur += 4;
|
|
// Save base context checksum
|
|
EncodeFixed32(cur, base_context_checksum);
|
|
cur += 4;
|
|
// Compute and save metaindex size
|
|
uint32_t metaindex_size = static_cast<uint32_t>(metaindex_handle.size());
|
|
if (metaindex_size != metaindex_handle.size()) {
|
|
return Status::NotSupported("Metaindex block size > 4GB");
|
|
}
|
|
// Metaindex must be adjacent to footer
|
|
assert(metaindex_size == 0 ||
|
|
metaindex_handle.offset() + metaindex_handle.size() ==
|
|
footer_offset - BlockTrailerSizeForMagicNumber(magic_number));
|
|
EncodeFixed32(cur, metaindex_size);
|
|
cur += 4;
|
|
|
|
// Zero pad remainder (for future use)
|
|
std::fill_n(cur, 24U, char{0});
|
|
assert(cur + 24 == part3);
|
|
|
|
// Compute checksum, add context
|
|
uint32_t checksum = ComputeBuiltinChecksum(
|
|
checksum_type, data_.data(), Footer::kNewVersionsEncodedLength);
|
|
checksum +=
|
|
ChecksumModifierForContext(base_context_checksum, footer_offset);
|
|
// Store it
|
|
EncodeFixed32(checksum_data, checksum);
|
|
} else {
|
|
// Base context checksum not used
|
|
assert(!FormatVersionUsesContextChecksum(format_version));
|
|
// Should be left empty
|
|
assert(base_context_checksum == 0);
|
|
assert(ChecksumModifierForContext(base_context_checksum, 0) == 0);
|
|
|
|
// Populate all of part 2
|
|
char* cur = part2;
|
|
cur = metaindex_handle.EncodeTo(cur);
|
|
cur = index_handle.EncodeTo(cur);
|
|
// Zero pad remainder
|
|
std::fill(cur, part3, char{0});
|
|
}
|
|
return Status::OK();
|
|
}
|
|
|
|
Status Footer::DecodeFrom(Slice input, uint64_t input_offset,
|
|
uint64_t enforce_table_magic_number) {
|
|
// Only decode to unused Footer
|
|
assert(table_magic_number_ == kNullTableMagicNumber);
|
|
assert(input != nullptr);
|
|
assert(input.size() >= kMinEncodedLength);
|
|
|
|
const char* magic_ptr = input.data() + input.size() - kMagicNumberLengthByte;
|
|
uint64_t magic = DecodeFixed64(magic_ptr);
|
|
|
|
// We check for legacy formats here and silently upconvert them
|
|
bool legacy = IsLegacyFooterFormat(magic);
|
|
if (legacy) {
|
|
magic = UpconvertLegacyFooterFormat(magic);
|
|
}
|
|
if (enforce_table_magic_number != 0 && enforce_table_magic_number != magic) {
|
|
return Status::Corruption("Bad table magic number: expected " +
|
|
std::to_string(enforce_table_magic_number) +
|
|
", found " + std::to_string(magic));
|
|
}
|
|
table_magic_number_ = magic;
|
|
block_trailer_size_ = BlockTrailerSizeForMagicNumber(magic);
|
|
|
|
// Parse Part3
|
|
const char* part3_ptr = magic_ptr;
|
|
uint32_t computed_checksum = 0;
|
|
uint64_t footer_offset = 0;
|
|
if (legacy) {
|
|
// The size is already asserted to be at least kMinEncodedLength
|
|
// at the beginning of the function
|
|
input.remove_prefix(input.size() - kVersion0EncodedLength);
|
|
format_version_ = 0 /* legacy */;
|
|
checksum_type_ = kCRC32c;
|
|
} else {
|
|
part3_ptr = magic_ptr - 4;
|
|
format_version_ = DecodeFixed32(part3_ptr);
|
|
if (UNLIKELY(!IsSupportedFormatVersion(format_version_) &&
|
|
!TEST_AllowUnsupportedFormatVersion())) {
|
|
return Status::Corruption("Corrupt or unsupported format_version: " +
|
|
std::to_string(format_version_));
|
|
}
|
|
// All known format versions >= 1 occupy exactly this many bytes.
|
|
if (UNLIKELY(input.size() < kNewVersionsEncodedLength)) {
|
|
return Status::Corruption("Input is too short to be an SST file");
|
|
}
|
|
uint64_t adjustment = input.size() - kNewVersionsEncodedLength;
|
|
input.remove_prefix(adjustment);
|
|
footer_offset = input_offset + adjustment;
|
|
|
|
// Parse Part1
|
|
char chksum = input.data()[0];
|
|
checksum_type_ = lossless_cast<ChecksumType>(chksum);
|
|
if (UNLIKELY(!IsSupportedChecksumType(checksum_type()))) {
|
|
return Status::Corruption("Corrupt or unsupported checksum type: " +
|
|
std::to_string(lossless_cast<uint8_t>(chksum)));
|
|
}
|
|
// This is the most convenient place to compute the checksum
|
|
if (checksum_type_ != kNoChecksum && format_version_ >= 6) {
|
|
std::array<char, kNewVersionsEncodedLength> copy_without_checksum;
|
|
std::copy_n(input.data(), kNewVersionsEncodedLength,
|
|
copy_without_checksum.data());
|
|
EncodeFixed32(©_without_checksum[5], 0); // Clear embedded checksum
|
|
computed_checksum =
|
|
ComputeBuiltinChecksum(checksum_type(), copy_without_checksum.data(),
|
|
kNewVersionsEncodedLength);
|
|
}
|
|
// Consume checksum type field
|
|
input.remove_prefix(1);
|
|
}
|
|
|
|
// Parse Part2
|
|
if (format_version_ >= 6) {
|
|
Slice ext_magic(input.data(), 4);
|
|
if (UNLIKELY(ext_magic.compare(Slice(kExtendedMagic.data(),
|
|
kExtendedMagic.size())) != 0)) {
|
|
return Status::Corruption("Bad extended magic number: 0x" +
|
|
ext_magic.ToString(/*hex*/ true));
|
|
}
|
|
input.remove_prefix(4);
|
|
uint32_t stored_checksum = 0, metaindex_size = 0;
|
|
bool success;
|
|
success = GetFixed32(&input, &stored_checksum);
|
|
assert(success);
|
|
success = GetFixed32(&input, &base_context_checksum_);
|
|
assert(success);
|
|
if (UNLIKELY(ChecksumModifierForContext(base_context_checksum_, 0) == 0)) {
|
|
return Status::Corruption("Invalid base context checksum");
|
|
}
|
|
computed_checksum +=
|
|
ChecksumModifierForContext(base_context_checksum_, footer_offset);
|
|
if (UNLIKELY(computed_checksum != stored_checksum)) {
|
|
return Status::Corruption("Footer at " + std::to_string(footer_offset) +
|
|
" checksum mismatch");
|
|
}
|
|
success = GetFixed32(&input, &metaindex_size);
|
|
assert(success);
|
|
(void)success;
|
|
uint64_t metaindex_end = footer_offset - GetBlockTrailerSize();
|
|
metaindex_handle_ =
|
|
BlockHandle(metaindex_end - metaindex_size, metaindex_size);
|
|
|
|
// Mark unpopulated
|
|
index_handle_ = BlockHandle::NullBlockHandle();
|
|
|
|
// 16 bytes of unchecked reserved padding
|
|
input.remove_prefix(16U);
|
|
|
|
// 8 bytes of checked reserved padding (expected to be zero unless using a
|
|
// future feature).
|
|
uint64_t reserved = 0;
|
|
success = GetFixed64(&input, &reserved);
|
|
assert(success);
|
|
if (UNLIKELY(reserved != 0)) {
|
|
return Status::NotSupported(
|
|
"File uses a future feature not supported in this version");
|
|
}
|
|
// End of part 2
|
|
assert(input.data() == part3_ptr);
|
|
} else {
|
|
// format_version_ < 6
|
|
Status result = metaindex_handle_.DecodeFrom(&input);
|
|
if (result.ok()) {
|
|
result = index_handle_.DecodeFrom(&input);
|
|
}
|
|
if (!result.ok()) {
|
|
return result;
|
|
}
|
|
// Padding in part2 is ignored
|
|
}
|
|
return Status::OK();
|
|
}
|
|
|
|
std::string Footer::ToString() const {
|
|
std::string result;
|
|
result.reserve(1024);
|
|
|
|
result.append("metaindex handle: " + metaindex_handle_.ToString() +
|
|
" offset: " + std::to_string(metaindex_handle_.offset()) +
|
|
" size: " + std::to_string(metaindex_handle_.size()) + "\n ");
|
|
result.append("index handle: " + index_handle_.ToString() +
|
|
" offset: " + std::to_string(index_handle_.offset()) +
|
|
" size: " + std::to_string(index_handle_.size()) + "\n ");
|
|
result.append("table_magic_number: " + std::to_string(table_magic_number_) +
|
|
"\n ");
|
|
if (!IsLegacyFooterFormat(table_magic_number_)) {
|
|
result.append("format version: " + std::to_string(format_version_) + "\n");
|
|
}
|
|
return result;
|
|
}
|
|
|
|
bool& TEST_AllowUnsupportedFormatVersion() {
|
|
static bool allow = false;
|
|
return allow;
|
|
}
|
|
|
|
static Status ReadFooterFromFileInternal(const IOOptions& opts,
|
|
RandomAccessFileReader* file,
|
|
FileSystem& fs,
|
|
FilePrefetchBuffer* prefetch_buffer,
|
|
uint64_t file_size, Footer* footer,
|
|
uint64_t enforce_table_magic_number) {
|
|
if (file_size < Footer::kMinEncodedLength) {
|
|
return Status::Corruption("file is too short (" +
|
|
std::to_string(file_size) +
|
|
" bytes) to be an "
|
|
"sstable: " +
|
|
file->file_name());
|
|
}
|
|
|
|
std::array<char, Footer::kMaxEncodedLength + 1> footer_buf;
|
|
AlignedBuf internal_buf;
|
|
Slice footer_input;
|
|
uint64_t read_offset = (file_size > Footer::kMaxEncodedLength)
|
|
? file_size - Footer::kMaxEncodedLength
|
|
: 0;
|
|
Status s;
|
|
// TODO: Need to pass appropriate deadline to TryReadFromCache(). Right now,
|
|
// there is no readahead for point lookups, so TryReadFromCache will fail if
|
|
// the required data is not in the prefetch buffer. Once deadline is enabled
|
|
// for iterator, TryReadFromCache might do a readahead. Revisit to see if we
|
|
// need to pass a timeout at that point
|
|
// TODO: rate limit footer reads.
|
|
if (prefetch_buffer == nullptr ||
|
|
!prefetch_buffer->TryReadFromCache(opts, file, read_offset,
|
|
Footer::kMaxEncodedLength,
|
|
&footer_input, nullptr)) {
|
|
if (file->use_direct_io()) {
|
|
s = file->Read(opts, read_offset, Footer::kMaxEncodedLength,
|
|
&footer_input, nullptr, &internal_buf);
|
|
} else {
|
|
s = file->Read(opts, read_offset, Footer::kMaxEncodedLength,
|
|
&footer_input, footer_buf.data(), nullptr);
|
|
}
|
|
if (!s.ok()) {
|
|
return s;
|
|
}
|
|
}
|
|
|
|
TEST_SYNC_POINT_CALLBACK("ReadFooterFromFileInternal:0", &footer_input);
|
|
|
|
// Check that we actually read the whole footer from the file. It may be
|
|
// that size isn't correct.
|
|
if (footer_input.size() < Footer::kMinEncodedLength) {
|
|
uint64_t size_on_disk = 0;
|
|
if (fs.GetFileSize(file->file_name(), IOOptions(), &size_on_disk, nullptr)
|
|
.ok()) {
|
|
// Similar to CheckConsistency message, but not completely sure the
|
|
// expected size always came from manifest.
|
|
return Status::Corruption("Sst file size mismatch: " + file->file_name() +
|
|
". Expected " + std::to_string(file_size) +
|
|
", actual size " +
|
|
std::to_string(size_on_disk) + "\n");
|
|
} else {
|
|
return Status::Corruption(
|
|
"Missing SST footer data in file " + file->file_name() +
|
|
" File too short? Expected size: " + std::to_string(file_size));
|
|
}
|
|
}
|
|
|
|
s = footer->DecodeFrom(footer_input, read_offset, enforce_table_magic_number);
|
|
if (!s.ok()) {
|
|
s = Status::CopyAppendMessage(s, " in ", file->file_name());
|
|
return s;
|
|
}
|
|
return Status::OK();
|
|
}
|
|
|
|
Status ReadFooterFromFile(const IOOptions& opts, RandomAccessFileReader* file,
|
|
FileSystem& fs, FilePrefetchBuffer* prefetch_buffer,
|
|
uint64_t file_size, Footer* footer,
|
|
uint64_t enforce_table_magic_number,
|
|
Statistics* stats) {
|
|
Status s =
|
|
ReadFooterFromFileInternal(opts, file, fs, prefetch_buffer, file_size,
|
|
footer, enforce_table_magic_number);
|
|
if (s.IsCorruption() &&
|
|
CheckFSFeatureSupport(&fs, FSSupportedOps::kVerifyAndReconstructRead)) {
|
|
IOOptions new_opts = opts;
|
|
new_opts.verify_and_reconstruct_read = true;
|
|
footer->Reset();
|
|
s = ReadFooterFromFileInternal(new_opts, file, fs,
|
|
/*prefetch_buffer=*/nullptr, file_size,
|
|
footer, enforce_table_magic_number);
|
|
RecordTick(stats, FILE_READ_CORRUPTION_RETRY_COUNT);
|
|
if (s.ok()) {
|
|
RecordTick(stats, FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT);
|
|
}
|
|
}
|
|
return s;
|
|
}
|
|
|
|
namespace {
|
|
// Custom handling for the last byte of a block, to avoid invoking streaming
|
|
// API to get an effective block checksum. This function is its own inverse
|
|
// because it uses xor.
|
|
inline uint32_t ModifyChecksumForLastByte(uint32_t checksum, char last_byte) {
|
|
// This strategy bears some resemblance to extending a CRC checksum by one
|
|
// more byte, except we don't need to re-mix the input checksum as long as
|
|
// we do this step only once (per checksum).
|
|
const uint32_t kRandomPrime = 0x6b9083d9;
|
|
return checksum ^ lossless_cast<uint8_t>(last_byte) * kRandomPrime;
|
|
}
|
|
} // namespace
|
|
|
|
uint32_t ComputeBuiltinChecksum(ChecksumType type, const char* data,
|
|
size_t data_size) {
|
|
switch (type) {
|
|
case kCRC32c:
|
|
return crc32c::Mask(crc32c::Value(data, data_size));
|
|
case kxxHash:
|
|
return XXH32(data, data_size, /*seed*/ 0);
|
|
case kxxHash64:
|
|
return Lower32of64(XXH64(data, data_size, /*seed*/ 0));
|
|
case kXXH3: {
|
|
if (data_size == 0) {
|
|
// Special case because of special handling for last byte, not
|
|
// present in this case. Can be any value different from other
|
|
// small input size checksums.
|
|
return 0;
|
|
} else {
|
|
// See corresponding code in ComputeBuiltinChecksumWithLastByte
|
|
uint32_t v = Lower32of64(XXH3_64bits(data, data_size - 1));
|
|
return ModifyChecksumForLastByte(v, data[data_size - 1]);
|
|
}
|
|
}
|
|
default: // including kNoChecksum
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
uint32_t ComputeBuiltinChecksumWithLastByte(ChecksumType type, const char* data,
|
|
size_t data_size, char last_byte) {
|
|
switch (type) {
|
|
case kCRC32c: {
|
|
uint32_t crc = crc32c::Value(data, data_size);
|
|
// Extend to cover last byte (compression type)
|
|
crc = crc32c::Extend(crc, &last_byte, 1);
|
|
return crc32c::Mask(crc);
|
|
}
|
|
case kxxHash: {
|
|
XXH32_state_t* const state = XXH32_createState();
|
|
XXH32_reset(state, 0);
|
|
XXH32_update(state, data, data_size);
|
|
// Extend to cover last byte (compression type)
|
|
XXH32_update(state, &last_byte, 1);
|
|
uint32_t v = XXH32_digest(state);
|
|
XXH32_freeState(state);
|
|
return v;
|
|
}
|
|
case kxxHash64: {
|
|
XXH64_state_t* const state = XXH64_createState();
|
|
XXH64_reset(state, 0);
|
|
XXH64_update(state, data, data_size);
|
|
// Extend to cover last byte (compression type)
|
|
XXH64_update(state, &last_byte, 1);
|
|
uint32_t v = Lower32of64(XXH64_digest(state));
|
|
XXH64_freeState(state);
|
|
return v;
|
|
}
|
|
case kXXH3: {
|
|
// XXH3 is a complicated hash function that is extremely fast on
|
|
// contiguous input, but that makes its streaming support rather
|
|
// complex. It is worth custom handling of the last byte (`type`)
|
|
// in order to avoid allocating a large state object and bringing
|
|
// that code complexity into CPU working set.
|
|
uint32_t v = Lower32of64(XXH3_64bits(data, data_size));
|
|
return ModifyChecksumForLastByte(v, last_byte);
|
|
}
|
|
default: // including kNoChecksum
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
Status DecompressBlockData(Decompressor::Args& args, Decompressor& decompressor,
|
|
BlockContents* out_contents,
|
|
const ImmutableOptions& ioptions,
|
|
MemoryAllocator* allocator) {
|
|
assert(args.compression_type != kNoCompression && "Invalid compression type");
|
|
|
|
StopWatchNano timer(ioptions.clock,
|
|
ShouldReportDetailedTime(ioptions.env, ioptions.stats));
|
|
|
|
Status s = decompressor.ExtractUncompressedSize(args);
|
|
if (UNLIKELY(!s.ok())) {
|
|
return s;
|
|
}
|
|
CacheAllocationPtr ubuf = AllocateBlock(args.uncompressed_size, allocator);
|
|
s = decompressor.DecompressBlock(args, ubuf.get());
|
|
if (UNLIKELY(!s.ok())) {
|
|
return s;
|
|
}
|
|
|
|
*out_contents = BlockContents(std::move(ubuf), args.uncompressed_size);
|
|
|
|
if (ShouldReportDetailedTime(ioptions.env, ioptions.stats)) {
|
|
RecordTimeToHistogram(ioptions.stats, DECOMPRESSION_TIMES_NANOS,
|
|
timer.ElapsedNanos());
|
|
}
|
|
RecordTick(ioptions.stats, BYTES_DECOMPRESSED_FROM,
|
|
args.compressed_data.size());
|
|
RecordTick(ioptions.stats, BYTES_DECOMPRESSED_TO, out_contents->data.size());
|
|
RecordTick(ioptions.stats, NUMBER_BLOCK_DECOMPRESSED);
|
|
|
|
TEST_SYNC_POINT_CALLBACK("DecompressBlockData:TamperWithReturnValue",
|
|
static_cast<void*>(&s));
|
|
TEST_SYNC_POINT_CALLBACK("DecompressBlockData:TamperWithDecompressionOutput",
|
|
static_cast<void*>(out_contents));
|
|
|
|
return s;
|
|
}
|
|
|
|
Status DecompressBlockData(const char* data, size_t size, CompressionType type,
|
|
Decompressor& decompressor,
|
|
BlockContents* out_contents,
|
|
const ImmutableOptions& ioptions,
|
|
MemoryAllocator* allocator,
|
|
Decompressor::ManagedWorkingArea* working_area) {
|
|
Decompressor::Args args;
|
|
args.compressed_data = Slice(data, size);
|
|
args.compression_type = type;
|
|
args.working_area = working_area;
|
|
return DecompressBlockData(args, decompressor, out_contents, ioptions,
|
|
allocator);
|
|
}
|
|
|
|
Status DecompressSerializedBlock(const char* data, size_t size,
|
|
CompressionType type,
|
|
Decompressor& decompressor,
|
|
BlockContents* out_contents,
|
|
const ImmutableOptions& ioptions,
|
|
MemoryAllocator* allocator) {
|
|
assert(data[size] != kNoCompression);
|
|
assert(data[size] == static_cast<char>(type));
|
|
return DecompressBlockData(data, size, type, decompressor, out_contents,
|
|
ioptions, allocator);
|
|
}
|
|
|
|
Status DecompressSerializedBlock(Decompressor::Args& args,
|
|
Decompressor& decompressor,
|
|
BlockContents* out_contents,
|
|
const ImmutableOptions& ioptions,
|
|
MemoryAllocator* allocator) {
|
|
assert(args.compressed_data.data()[args.compressed_data.size()] !=
|
|
kNoCompression);
|
|
assert(args.compressed_data.data()[args.compressed_data.size()] ==
|
|
static_cast<char>(args.compression_type));
|
|
return DecompressBlockData(args, decompressor, out_contents, ioptions,
|
|
allocator);
|
|
}
|
|
|
|
// Replace the contents of db_host_id with the actual hostname, if db_host_id
|
|
// matches the keyword kHostnameForDbHostId
|
|
Status ReifyDbHostIdProperty(Env* env, std::string* db_host_id) {
|
|
assert(db_host_id);
|
|
if (*db_host_id == kHostnameForDbHostId) {
|
|
Status s = env->GetHostNameString(db_host_id);
|
|
if (!s.ok()) {
|
|
db_host_id->clear();
|
|
}
|
|
return s;
|
|
}
|
|
|
|
return Status::OK();
|
|
}
|
|
} // namespace ROCKSDB_NAMESPACE
|