rocksdb/table/format.cc
Peter Dillinger 9d490593d0 Preliminary support for custom compression algorithms (#13659)
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
2025-06-16 14:19:03 -07:00

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(&copy_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