Compare commits

..
Author SHA1 Message Date
Siying Dong 6ad033ce2b Fix test name typo in PlainTableDBTest 2019-01-25 17:31:06 -08:00
56 changed files with 576 additions and 804 deletions

No files matched your search

-2
View File
@@ -6,7 +6,6 @@
* RocksDB may choose to preopen some files even if options.max_open_files != -1. This may make DB open slightly longer.
* For users of dictionary compression with ZSTD v0.7.0+, we now reuse the same digested dictionary when compressing each of an SST file's data blocks for faster compression speeds.
* For all users of dictionary compression who set `cache_index_and_filter_blocks == true`, we now store dictionary data used for decompression in the block cache for better control over memory usage. For users of ZSTD v1.1.4+ who compile with -DZSTD_STATIC_LINKING_ONLY, this includes a digested dictionary, which is used to increase decompression speed.
* Add support for block checksums verification for external SST files before ingestion.
### Public API Change
* CompactionPri = kMinOverlappingRatio also uses compensated file size, which boosts file with lots of tombstones to be compacted first.
@@ -16,7 +15,6 @@
* With "ldb ----try_load_options", when wal_dir specified by the option file doesn't exist, ignore it.
* Change time resolution in FileOperationInfo.
* Deleting Blob files also go through SStFileManager.
* Remove PlainTable's store_index_in_file feature. When opening an existing DB with index in SST files, the index and bloom filter will still be rebuild while SST files are opened, in the same way as there is no index in the file.
### Bug Fixes
* Fix a deadlock caused by compaction and file ingestion waiting for each other in the event of write stalls.
+3 -2
View File
@@ -817,6 +817,7 @@ J ?= 100%
# Use this regexp to select the subset of tests whose names match.
tests-regexp = .
t_run = $(wildcard t/run-*)
.PHONY: check_0
check_0:
$(AM_V_GEN)export TEST_TMPDIR=$(TMPD); \
@@ -826,7 +827,7 @@ check_0:
test -t 1 && eta=--eta || eta=; \
{ \
printf './%s\n' $(filter-out $(PARALLEL_TEST),$(TESTS)); \
find t -name 'run-*' -print; \
printf '%s\n' $(t_run); \
} \
| $(prioritize_long_running_tests) \
| grep -E '$(tests-regexp)' \
@@ -843,7 +844,7 @@ valgrind_check_0:
test -t 1 && eta=--eta || eta=; \
{ \
printf './%s\n' $(filter-out $(PARALLEL_TEST) %skiplist_test options_settable_test, $(TESTS)); \
find t -name 'run-*' -print; \
printf '%s\n' $(t_run); \
} \
| $(prioritize_long_running_tests) \
| grep -E '$(tests-regexp)' \
+3 -5
View File
@@ -53,13 +53,11 @@ if [ -z "$ROCKSDB_NO_FBCODE" -a -d /mnt/gvfs/third-party ]; then
FBCODE_BUILD="true"
# If we're compiling with TSAN we need pic build
PIC_BUILD=$COMPILE_WITH_TSAN
if [ -n "$ROCKSDB_FBCODE_BUILD_WITH_481" ]; then
if [ -z "$ROCKSDB_FBCODE_BUILD_WITH_481" ]; then
source "$PWD/build_tools/fbcode_config.sh"
else
# we need this to build with MySQL. Don't use for other purposes.
source "$PWD/build_tools/fbcode_config4.8.1.sh"
elif [ -n "$ROCKSDB_FBCODE_BUILD_WITH_PLATFORM007" ]; then
source "$PWD/build_tools/fbcode_config_platform007.sh"
else
source "$PWD/build_tools/fbcode_config.sh"
fi
fi
-18
View File
@@ -1,18 +0,0 @@
GCC_BASE=/mnt/gvfs/third-party2/gcc/6e8e715624fd15256a7970073387793dfcf79b46/7.x/centos7-native/b2ef2b6
CLANG_BASE=/mnt/gvfs/third-party2/llvm-fb/ef37e1faa1c29782abfac1ae65a291b9b7966f6d/stable/centos7-native/c9f9104
LIBGCC_BASE=/mnt/gvfs/third-party2/libgcc/c67031f0f739ac61575a061518d6ef5038f99f90/7.x/platform007/5620abc
GLIBC_BASE=/mnt/gvfs/third-party2/glibc/60d6f124a78798b73944f5ba87c2306ae3460153/2.26/platform007/f259413
SNAPPY_BASE=/mnt/gvfs/third-party2/snappy/7f9bdaada18f59bc27ec2b0871eb8a6144343aef/1.1.3/platform007/ca4da3d
ZLIB_BASE=/mnt/gvfs/third-party2/zlib/22c2d65676fb7c23cfa797c4f6937f38b026f3cf/1.2.8/platform007/ca4da3d
BZIP2_BASE=/mnt/gvfs/third-party2/bzip2/dc49a21c5fceec6456a7a28a94dcd16690af1337/1.0.6/platform007/ca4da3d
LZ4_BASE=/mnt/gvfs/third-party2/lz4/907b498203d297947f3bb70b9466f47e100f1873/r131/platform007/ca4da3d
ZSTD_BASE=/mnt/gvfs/third-party2/zstd/3ee276cbacfad3074e3f07bf826ac47f06970f4e/1.3.5/platform007/15a3614
GFLAGS_BASE=/mnt/gvfs/third-party2/gflags/0b9929d2588991c65a57168bf88aff2db87c5d48/2.2.0/platform007/ca4da3d
JEMALLOC_BASE=/mnt/gvfs/third-party2/jemalloc/9c910d36d6235cc40e8ff559358f1833452300ca/master/platform007/5b0f53e
NUMA_BASE=/mnt/gvfs/third-party2/numa/9cbf2460284c669ed19c3ccb200a71f7dd7e53c7/2.0.11/platform007/ca4da3d
LIBUNWIND_BASE=/mnt/gvfs/third-party2/libunwind/bf3d7497fe4e6d007354f0adffa16ce3003f8338/1.3/platform007/6f3e0a9
TBB_BASE=/mnt/gvfs/third-party2/tbb/ff4e0b093534704d8abab678a4fd7f5ea7b094c7/2018_U5/platform007/ca4da3d
KERNEL_HEADERS_BASE=/mnt/gvfs/third-party2/kernel-headers/b5c4a61a5c483ba24722005ae07895971a2ac707/fb/platform007/da39a3e
BINUTILS_BASE=/mnt/gvfs/third-party2/binutils/92ff90349e2f43ea0a8246d8b1cf17b6869013e3/2.29.1/centos7-native/da39a3e
VALGRIND_BASE=/mnt/gvfs/third-party2/valgrind/f3f697a28122e6bcd513273dd9c1ff23852fc59f/3.13.0/platform007/ca4da3d
LUA_BASE=/mnt/gvfs/third-party2/lua/f0cd714433206d5139df61659eb7b28b1dea6683/5.3.4/platform007/5007832
-157
View File
@@ -1,157 +0,0 @@
#!/bin/sh
#
# Set environment variables so that we can compile rocksdb using
# fbcode settings. It uses the latest g++ and clang compilers and also
# uses jemalloc
# Environment variables that change the behavior of this script:
# PIC_BUILD -- if true, it will only take pic versions of libraries from fbcode. libraries that don't have pic variant will not be included
BASEDIR=`dirname $BASH_SOURCE`
source "$BASEDIR/dependencies_platform007.sh"
CFLAGS=""
# libgcc
LIBGCC_INCLUDE="$LIBGCC_BASE/include/c++/7.3.0"
LIBGCC_LIBS=" -L $LIBGCC_BASE/lib"
# glibc
GLIBC_INCLUDE="$GLIBC_BASE/include"
GLIBC_LIBS=" -L $GLIBC_BASE/lib"
# snappy
SNAPPY_INCLUDE=" -I $SNAPPY_BASE/include/"
if test -z $PIC_BUILD; then
SNAPPY_LIBS=" $SNAPPY_BASE/lib/libsnappy.a"
else
SNAPPY_LIBS=" $SNAPPY_BASE/lib/libsnappy_pic.a"
fi
CFLAGS+=" -DSNAPPY"
if test -z $PIC_BUILD; then
# location of zlib headers and libraries
ZLIB_INCLUDE=" -I $ZLIB_BASE/include/"
ZLIB_LIBS=" $ZLIB_BASE/lib/libz.a"
CFLAGS+=" -DZLIB"
# location of bzip headers and libraries
BZIP_INCLUDE=" -I $BZIP2_BASE/include/"
BZIP_LIBS=" $BZIP2_BASE/lib/libbz2.a"
CFLAGS+=" -DBZIP2"
LZ4_INCLUDE=" -I $LZ4_BASE/include/"
LZ4_LIBS=" $LZ4_BASE/lib/liblz4.a"
CFLAGS+=" -DLZ4"
fi
ZSTD_INCLUDE=" -I $ZSTD_BASE/include/"
if test -z $PIC_BUILD; then
ZSTD_LIBS=" $ZSTD_BASE/lib/libzstd.a"
else
ZSTD_LIBS=" $ZSTD_BASE/lib/libzstd_pic.a"
fi
CFLAGS+=" -DZSTD"
# location of gflags headers and libraries
GFLAGS_INCLUDE=" -I $GFLAGS_BASE/include/"
if test -z $PIC_BUILD; then
GFLAGS_LIBS=" $GFLAGS_BASE/lib/libgflags.a"
else
GFLAGS_LIBS=" $GFLAGS_BASE/lib/libgflags_pic.a"
fi
CFLAGS+=" -DGFLAGS=gflags"
# location of jemalloc
JEMALLOC_INCLUDE=" -I $JEMALLOC_BASE/include/"
JEMALLOC_LIB=" $JEMALLOC_BASE/lib/libjemalloc.a"
if test -z $PIC_BUILD; then
# location of numa
NUMA_INCLUDE=" -I $NUMA_BASE/include/"
NUMA_LIB=" $NUMA_BASE/lib/libnuma.a"
CFLAGS+=" -DNUMA"
# location of libunwind
LIBUNWIND="$LIBUNWIND_BASE/lib/libunwind.a"
fi
# location of TBB
TBB_INCLUDE=" -isystem $TBB_BASE/include/"
if test -z $PIC_BUILD; then
TBB_LIBS="$TBB_BASE/lib/libtbb.a"
else
TBB_LIBS="$TBB_BASE/lib/libtbb_pic.a"
fi
CFLAGS+=" -DTBB"
# use Intel SSE support for checksum calculations
export USE_SSE=1
export PORTABLE=1
BINUTILS="$BINUTILS_BASE/bin"
AR="$BINUTILS/ar"
DEPS_INCLUDE="$SNAPPY_INCLUDE $ZLIB_INCLUDE $BZIP_INCLUDE $LZ4_INCLUDE $ZSTD_INCLUDE $GFLAGS_INCLUDE $NUMA_INCLUDE $TBB_INCLUDE"
STDLIBS="-L $GCC_BASE/lib64"
CLANG_BIN="$CLANG_BASE/bin"
CLANG_LIB="$CLANG_BASE/lib"
CLANG_SRC="$CLANG_BASE/../../src"
CLANG_ANALYZER="$CLANG_BIN/clang++"
CLANG_SCAN_BUILD="$CLANG_SRC/llvm/tools/clang/tools/scan-build/bin/scan-build"
if [ -z "$USE_CLANG" ]; then
# gcc
CC="$GCC_BASE/bin/gcc"
CXX="$GCC_BASE/bin/g++"
CFLAGS+=" -B$BINUTILS/gold"
CFLAGS+=" -isystem $LIBGCC_INCLUDE"
CFLAGS+=" -isystem $GLIBC_INCLUDE"
JEMALLOC=1
else
# clang
CLANG_INCLUDE="$CLANG_LIB/clang/stable/include"
CC="$CLANG_BIN/clang"
CXX="$CLANG_BIN/clang++"
KERNEL_HEADERS_INCLUDE="$KERNEL_HEADERS_BASE/include"
CFLAGS+=" -B$BINUTILS/gold -nostdinc -nostdlib"
CFLAGS+=" -isystem $LIBGCC_BASE/include/c++/7.x "
CFLAGS+=" -isystem $LIBGCC_BASE/include/c++/7.x/x86_64-facebook-linux "
CFLAGS+=" -isystem $GLIBC_INCLUDE"
CFLAGS+=" -isystem $LIBGCC_INCLUDE"
CFLAGS+=" -isystem $CLANG_INCLUDE"
CFLAGS+=" -isystem $KERNEL_HEADERS_INCLUDE/linux "
CFLAGS+=" -isystem $KERNEL_HEADERS_INCLUDE "
CFLAGS+=" -Wno-expansion-to-defined "
CXXFLAGS="-nostdinc++"
fi
CFLAGS+=" $DEPS_INCLUDE"
CFLAGS+=" -DROCKSDB_PLATFORM_POSIX -DROCKSDB_LIB_IO_POSIX -DROCKSDB_FALLOCATE_PRESENT -DROCKSDB_MALLOC_USABLE_SIZE -DROCKSDB_RANGESYNC_PRESENT -DROCKSDB_SCHED_GETCPU_PRESENT -DROCKSDB_SUPPORT_THREAD_LOCAL -DHAVE_SSE42"
CXXFLAGS+=" $CFLAGS"
EXEC_LDFLAGS=" $SNAPPY_LIBS $ZLIB_LIBS $BZIP_LIBS $LZ4_LIBS $ZSTD_LIBS $GFLAGS_LIBS $NUMA_LIB $TBB_LIBS"
EXEC_LDFLAGS+=" -B$BINUTILS/gold"
EXEC_LDFLAGS+=" -Wl,--dynamic-linker,/usr/local/fbcode/platform007/lib/ld.so"
EXEC_LDFLAGS+=" $LIBUNWIND"
EXEC_LDFLAGS+=" -Wl,-rpath=/usr/local/fbcode/platform007/lib"
# required by libtbb
EXEC_LDFLAGS+=" -ldl"
PLATFORM_LDFLAGS="$LIBGCC_LIBS $GLIBC_LIBS $STDLIBS -lgcc -lstdc++"
EXEC_LDFLAGS_SHARED="$SNAPPY_LIBS $ZLIB_LIBS $BZIP_LIBS $LZ4_LIBS $ZSTD_LIBS $GFLAGS_LIBS $TBB_LIBS"
VALGRIND_VER="$VALGRIND_BASE/bin/"
# lua not supported because it's on track for deprecation, I think
LUA_PATH=
LUA_LIB=
export CC CXX AR CFLAGS CXXFLAGS EXEC_LDFLAGS EXEC_LDFLAGS_SHARED VALGRIND_VER JEMALLOC_LIB JEMALLOC_INCLUDE CLANG_ANALYZER CLANG_SCAN_BUILD LUA_PATH LUA_LIB
+1 -3
View File
@@ -85,9 +85,7 @@ NON_SHM="TMPD=/tmp/rocksdb_test_tmp"
GCC_481="ROCKSDB_FBCODE_BUILD_WITH_481=1"
ASAN="COMPILE_WITH_ASAN=1"
CLANG="USE_CLANG=1"
# in gcc-5 there are known problems with TSAN like https://gcc.gnu.org/bugzilla/show_bug.cgi?id=71090.
# using platform007 gives us gcc-8 or higher which has that bug fixed.
TSAN="ROCKSDB_FBCODE_BUILD_WITH_PLATFORM007=1 COMPILE_WITH_TSAN=1"
TSAN="COMPILE_WITH_TSAN=1"
UBSAN="COMPILE_WITH_UBSAN=1"
TSAN_CRASH='CRASH_TEST_EXT_ARGS="--compression_type=zstd --log2_keys_per_lock=22"'
NON_TSAN_CRASH="CRASH_TEST_EXT_ARGS=--compression_type=zstd"
-39
View File
@@ -53,45 +53,6 @@ function get_lib_base()
log_variable $__res_var
}
###########################################################
# platform007 dependencies #
###########################################################
OUTPUT="$BASEDIR/dependencies_platform007.sh"
rm -f "$OUTPUT"
touch "$OUTPUT"
echo "Writing dependencies to $OUTPUT"
# Compilers locations
GCC_BASE=`readlink -f $TP2_LATEST/gcc/7.x/centos7-native/*/`
CLANG_BASE=`readlink -f $TP2_LATEST/llvm-fb/stable/centos7-native/*/`
log_variable GCC_BASE
log_variable CLANG_BASE
# Libraries locations
get_lib_base libgcc 7.x platform007
get_lib_base glibc 2.26 platform007
get_lib_base snappy LATEST platform007
get_lib_base zlib LATEST platform007
get_lib_base bzip2 LATEST platform007
get_lib_base lz4 LATEST platform007
get_lib_base zstd LATEST platform007
get_lib_base gflags LATEST platform007
get_lib_base jemalloc LATEST platform007
get_lib_base numa LATEST platform007
get_lib_base libunwind LATEST platform007
get_lib_base tbb LATEST platform007
get_lib_base kernel-headers fb platform007
get_lib_base binutils LATEST centos7-native
get_lib_base valgrind LATEST platform007
get_lib_base lua 5.3.4 platform007
git diff $OUTPUT
###########################################################
# 5.x dependencies #
###########################################################
+3 -3
View File
@@ -121,9 +121,9 @@ Status BuildTable(
file->SetIOPriority(io_priority);
file->SetWriteLifeTimeHint(write_hint);
file_writer.reset(
new WritableFileWriter(std::move(file), fname, env_options, env,
ioptions.statistics, ioptions.listeners));
file_writer.reset(new WritableFileWriter(std::move(file), fname,
env_options, ioptions.statistics,
ioptions.listeners));
builder = NewTableBuilder(
ioptions, mutable_cf_options, internal_comparator,
int_tbl_prop_collector_factories, column_family_id,
+7 -26
View File
@@ -604,14 +604,7 @@ Status CompactionJob::Run() {
}
compaction_stats_.micros = env_->NowMicros() - start_micros;
compaction_stats_.cpu_micros = 0;
for (size_t i = 0; i < compact_->sub_compact_states.size(); i++) {
compaction_stats_.cpu_micros +=
compact_->sub_compact_states[i].compaction_job_stats.cpu_micros;
}
MeasureTime(stats_, COMPACTION_TIME, compaction_stats_.micros);
MeasureTime(stats_, COMPACTION_CPU_TIME, compaction_stats_.cpu_micros);
TEST_SYNC_POINT("CompactionJob::Run:BeforeVerify");
@@ -774,7 +767,6 @@ Status CompactionJob::Install(const MutableCFOptions& mutable_cf_options) {
stream << "job" << job_id_ << "event"
<< "compaction_finished"
<< "compaction_time_micros" << compaction_stats_.micros
<< "compaction_time_cpu_micros" << compaction_stats_.cpu_micros
<< "output_level" << compact_->compaction->output_level()
<< "num_output_files" << compact_->NumOutputFiles()
<< "total_output_size" << compact_->total_bytes << "num_input_records"
@@ -812,9 +804,6 @@ Status CompactionJob::Install(const MutableCFOptions& mutable_cf_options) {
void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
assert(sub_compact != nullptr);
uint64_t prev_cpu_micros = env_->NowCPUNanos() / 1000;
ColumnFamilyData* cfd = sub_compact->compaction->column_family_data();
CompactionRangeDelAggregator range_del_agg(&cfd->internal_comparator(),
existing_snapshots_);
@@ -834,17 +823,13 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
uint64_t prev_fsync_nanos = 0;
uint64_t prev_range_sync_nanos = 0;
uint64_t prev_prepare_write_nanos = 0;
uint64_t prev_cpu_write_nanos = 0;
uint64_t prev_cpu_read_nanos = 0;
if (measure_io_stats_) {
prev_perf_level = GetPerfLevel();
SetPerfLevel(PerfLevel::kEnableTimeAndCPUTimeExceptForMutex);
SetPerfLevel(PerfLevel::kEnableTime);
prev_write_nanos = IOSTATS(write_nanos);
prev_fsync_nanos = IOSTATS(fsync_nanos);
prev_range_sync_nanos = IOSTATS(range_sync_nanos);
prev_prepare_write_nanos = IOSTATS(prepare_write_nanos);
prev_cpu_write_nanos = IOSTATS(cpu_write_nanos);
prev_cpu_read_nanos = IOSTATS(cpu_read_nanos);
}
const MutableCFOptions* mutable_cf_options =
@@ -1122,9 +1107,6 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
RecordDroppedKeys(range_del_out_stats, &sub_compact->compaction_job_stats);
}
sub_compact->compaction_job_stats.cpu_micros =
env_->NowCPUNanos() / 1000 - prev_cpu_micros;
if (measure_io_stats_) {
sub_compact->compaction_job_stats.file_write_nanos +=
IOSTATS(write_nanos) - prev_write_nanos;
@@ -1134,11 +1116,7 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
IOSTATS(range_sync_nanos) - prev_range_sync_nanos;
sub_compact->compaction_job_stats.file_prepare_write_nanos +=
IOSTATS(prepare_write_nanos) - prev_prepare_write_nanos;
sub_compact->compaction_job_stats.cpu_micros -=
(IOSTATS(cpu_write_nanos) - prev_cpu_write_nanos +
IOSTATS(cpu_read_nanos) - prev_cpu_read_nanos) /
1000;
if (prev_perf_level != PerfLevel::kEnableTimeAndCPUTimeExceptForMutex) {
if (prev_perf_level != PerfLevel::kEnableTime) {
SetPerfLevel(prev_perf_level);
}
}
@@ -1427,7 +1405,10 @@ Status CompactionJob::FinishCompactionOutputFile(
auto sfm =
static_cast<SstFileManagerImpl*>(db_options_.sst_file_manager.get());
if (sfm && meta != nullptr && meta->fd.GetPathId() == 0) {
sfm->OnAddFile(fname);
auto fn =
TableFileName(sub_compact->compaction->immutable_cf_options()->cf_paths,
meta->fd.GetNumber(), meta->fd.GetPathId());
sfm->OnAddFile(fn);
if (sfm->IsMaxAllowedSpaceReached()) {
// TODO(ajkr): should we return OK() if max space was reached by the final
// compaction output file (similarly to how flush works when full)?
@@ -1549,7 +1530,7 @@ Status CompactionJob::OpenCompactionOutputFile(
sub_compact->compaction->immutable_cf_options()->listeners;
sub_compact->outfile.reset(
new WritableFileWriter(std::move(writable_file), fname, env_options_,
env_, db_options_.statistics.get(), listeners));
db_options_.statistics.get(), listeners));
// If the Column family flag is to only optimize filters for hits,
// we can skip creating filters if this is the bottommost_level where
+5 -7
View File
@@ -923,13 +923,11 @@ class DBImpl : public DB {
// Flush the in-memory write buffer to storage. Switches to a new
// log-file/memtable and writes a new descriptor iff successful. Then
// installs a new super version for the column family.
Status FlushMemTableToOutputFile(
ColumnFamilyData* cfd, const MutableCFOptions& mutable_cf_options,
bool* madeProgress, JobContext* job_context,
SuperVersionContext* superversion_context,
std::vector<SequenceNumber>& snapshot_seqs,
SequenceNumber earliest_write_conflict_snapshot,
SnapshotChecker* snapshot_checker, LogBuffer* log_buffer);
Status FlushMemTableToOutputFile(ColumnFamilyData* cfd,
const MutableCFOptions& mutable_cf_options,
bool* madeProgress, JobContext* job_context,
SuperVersionContext* superversion_context,
LogBuffer* log_buffer);
// Argument required by background flush thread.
struct BGFlushArg {
+9 -13
View File
@@ -132,14 +132,16 @@ Status DBImpl::SyncClosedLogs(JobContext* job_context) {
Status DBImpl::FlushMemTableToOutputFile(
ColumnFamilyData* cfd, const MutableCFOptions& mutable_cf_options,
bool* made_progress, JobContext* job_context,
SuperVersionContext* superversion_context,
std::vector<SequenceNumber>& snapshot_seqs,
SequenceNumber earliest_write_conflict_snapshot,
SnapshotChecker* snapshot_checker, LogBuffer* log_buffer) {
SuperVersionContext* superversion_context, LogBuffer* log_buffer) {
mutex_.AssertHeld();
assert(cfd->imm()->NumNotFlushed() != 0);
assert(cfd->imm()->IsFlushPending());
std::vector<SequenceNumber> snapshot_seqs;
SequenceNumber earliest_write_conflict_snapshot;
SnapshotChecker* snapshot_checker;
GetSnapshotContext(job_context, &snapshot_seqs,
&earliest_write_conflict_snapshot, &snapshot_checker);
FlushJob flush_job(
dbname_, cfd, immutable_db_options_, mutable_cf_options,
@@ -237,20 +239,14 @@ Status DBImpl::FlushMemTablesToOutputFiles(
return AtomicFlushMemTablesToOutputFiles(bg_flush_args, made_progress,
job_context, log_buffer);
}
std::vector<SequenceNumber> snapshot_seqs;
SequenceNumber earliest_write_conflict_snapshot;
SnapshotChecker* snapshot_checker;
GetSnapshotContext(job_context, &snapshot_seqs,
&earliest_write_conflict_snapshot, &snapshot_checker);
Status status;
for (auto& arg : bg_flush_args) {
ColumnFamilyData* cfd = arg.cfd_;
MutableCFOptions mutable_cf_options = *cfd->GetLatestMutableCFOptions();
SuperVersionContext* superversion_context = arg.superversion_context_;
Status s = FlushMemTableToOutputFile(
cfd, mutable_cf_options, made_progress, job_context,
superversion_context, snapshot_seqs, earliest_write_conflict_snapshot,
snapshot_checker, log_buffer);
Status s = FlushMemTableToOutputFile(cfd, mutable_cf_options, made_progress,
job_context, superversion_context,
log_buffer);
if (!s.ok()) {
status = s;
if (!s.IsShutdownInProgress()) {
+2 -2
View File
@@ -237,7 +237,7 @@ Status DBImpl::NewDB() {
file->SetPreallocationBlockSize(
immutable_db_options_.manifest_preallocation_size);
std::unique_ptr<WritableFileWriter> file_writer(new WritableFileWriter(
std::move(file), manifest, env_options, env_, nullptr /* stats */,
std::move(file), manifest, env_options, nullptr /* stats */,
immutable_db_options_.listeners));
log::Writer log(std::move(file_writer), 0, false);
std::string record;
@@ -1176,7 +1176,7 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
const auto& listeners = impl->immutable_db_options_.listeners;
std::unique_ptr<WritableFileWriter> file_writer(
new WritableFileWriter(std::move(lfile), log_fname, opt_env_options,
impl->env_, nullptr /* stats */, listeners));
nullptr /* stats */, listeners));
impl->logs_.emplace_back(
new_log_number,
new log::Writer(
+1 -1
View File
@@ -1456,7 +1456,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
lfile->SetPreallocationBlockSize(preallocate_block_size);
lfile->SetWriteLifeTimeHint(write_hint);
std::unique_ptr<WritableFileWriter> file_writer(new WritableFileWriter(
std::move(lfile), log_fname, opt_env_opt, env_, nullptr /* stats */,
std::move(lfile), log_fname, opt_env_opt, nullptr /* stats */,
immutable_db_options_.listeners));
new_log = new log::Writer(
std::move(file_writer), new_log_number,
+1 -1
View File
@@ -1763,7 +1763,7 @@ TEST_F(DBTest2, TestPerfContextCpuTime) {
SetPerfLevel(PerfLevel::kEnableTimeAndCPUTimeExceptForMutex);
ASSERT_EQ("bar", Get("foo"));
ASSERT_GT(env_->now_cpu_count_.load(), 2);
ASSERT_EQ(env_->now_cpu_count_.load(), 2);
ASSERT_LT(get_perf_context()->get_cpu_nanos, kDummyAddonTime);
ASSERT_GT(get_perf_context()->find_table_nanos, kDummyAddonTime);
+122 -237
View File
@@ -14,9 +14,8 @@
namespace rocksdb {
#ifndef ROCKSDB_LITE
class ExternalSSTFileBasicTest
: public DBTestBase,
public ::testing::WithParamInterface<std::tuple<bool, bool>> {
class ExternalSSTFileBasicTest : public DBTestBase,
public ::testing::WithParamInterface<bool> {
public:
ExternalSSTFileBasicTest() : DBTestBase("/external_sst_file_basic_test") {
sst_files_dir_ = dbname_ + "/sst_files/";
@@ -43,8 +42,7 @@ class ExternalSSTFileBasicTest
const Options options, std::vector<int> keys,
const std::vector<ValueType>& value_types,
std::vector<std::pair<int, int>> range_deletions, int file_id,
bool write_global_seqno, bool verify_checksums_before_ingest,
std::map<std::string, std::string>* true_data) {
bool write_global_seqno, std::map<std::string, std::string>* true_data) {
assert(value_types.size() == 1 || keys.size() == value_types.size());
std::string file_path = sst_files_dir_ + ToString(file_id);
SstFileWriter sst_file_writer(EnvOptions(), options);
@@ -109,7 +107,6 @@ class ExternalSSTFileBasicTest
IngestExternalFileOptions ifo;
ifo.allow_global_seqno = true;
ifo.write_global_seqno = write_global_seqno;
ifo.verify_checksums_before_ingest = verify_checksums_before_ingest;
s = db_->IngestExternalFile({file_path}, ifo);
}
return s;
@@ -118,20 +115,18 @@ class ExternalSSTFileBasicTest
Status GenerateAndAddExternalFile(
const Options options, std::vector<int> keys,
const std::vector<ValueType>& value_types, int file_id,
bool write_global_seqno, bool verify_checksums_before_ingest,
std::map<std::string, std::string>* true_data) {
return GenerateAndAddExternalFile(
options, keys, value_types, {}, file_id, write_global_seqno,
verify_checksums_before_ingest, true_data);
bool write_global_seqno, std::map<std::string, std::string>* true_data) {
return GenerateAndAddExternalFile(options, keys, value_types, {}, file_id,
write_global_seqno, true_data);
}
Status GenerateAndAddExternalFile(
const Options options, std::vector<int> keys, const ValueType value_type,
int file_id, bool write_global_seqno, bool verify_checksums_before_ingest,
int file_id, bool write_global_seqno,
std::map<std::string, std::string>* true_data) {
return GenerateAndAddExternalFile(
options, keys, std::vector<ValueType>(1, value_type), file_id,
write_global_seqno, verify_checksums_before_ingest, true_data);
return GenerateAndAddExternalFile(options, keys,
std::vector<ValueType>(1, value_type),
file_id, write_global_seqno, true_data);
}
~ExternalSSTFileBasicTest() { test::DestroyDir(env_, sst_files_dir_); }
@@ -254,8 +249,7 @@ TEST_F(ExternalSSTFileBasicTest, NoCopy) {
}
TEST_P(ExternalSSTFileBasicTest, IngestFileWithGlobalSeqnoPickedSeqno) {
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
bool write_global_seqno = GetParam();
do {
Options options = CurrentOptions();
DestroyAndReopen(options);
@@ -263,39 +257,39 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithGlobalSeqnoPickedSeqno) {
int file_id = 1;
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 2, 3, 4, 5, 6}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 2, 3, 4, 5, 6},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 0);
ASSERT_OK(GenerateAndAddExternalFile(
options, {10, 11, 12, 13}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {10, 11, 12, 13},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 0);
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 4, 6}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 4, 6},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 1);
ASSERT_OK(GenerateAndAddExternalFile(
options, {11, 15, 19}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {11, 15, 19},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 2);
ASSERT_OK(GenerateAndAddExternalFile(
options, {120, 130}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {120, 130},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 2);
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 130}, ValueType::kTypeValue, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 130},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 3);
@@ -306,21 +300,21 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithGlobalSeqnoPickedSeqno) {
}
SequenceNumber last_seqno = dbfull()->GetLatestSequenceNumber();
ASSERT_OK(GenerateAndAddExternalFile(
options, {60, 61, 62}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {60, 61, 62},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno);
ASSERT_OK(GenerateAndAddExternalFile(
options, {40, 41, 42}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {40, 41, 42},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 1);
ASSERT_OK(GenerateAndAddExternalFile(
options, {20, 30, 40}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {20, 30, 40},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 2);
@@ -328,29 +322,29 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithGlobalSeqnoPickedSeqno) {
// We will need a seqno for the file regardless if the file overwrite
// keys in the DB or not because we have a snapshot
ASSERT_OK(GenerateAndAddExternalFile(
options, {1000, 1002}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1000, 1002},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 3);
ASSERT_OK(GenerateAndAddExternalFile(
options, {2000, 3002}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {2000, 3002},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 4);
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 20, 40, 100, 150}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 20, 40, 100, 150},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 5);
db_->ReleaseSnapshot(snapshot);
ASSERT_OK(GenerateAndAddExternalFile(
options, {5000, 5001}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {5000, 5001},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// No snapshot anymore, no need to assign a seqno
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 5);
@@ -360,8 +354,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithGlobalSeqnoPickedSeqno) {
}
TEST_P(ExternalSSTFileBasicTest, IngestFileWithMultipleValueType) {
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
bool write_global_seqno = GetParam();
do {
Options options = CurrentOptions();
options.merge_operator.reset(new TestPutOperator());
@@ -370,59 +363,59 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMultipleValueType) {
int file_id = 1;
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 2, 3, 4, 5, 6}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 2, 3, 4, 5, 6},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 0);
ASSERT_OK(GenerateAndAddExternalFile(
options, {10, 11, 12, 13}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {10, 11, 12, 13},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 0);
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 4, 6}, ValueType::kTypeMerge, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 4, 6},
ValueType::kTypeMerge, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 1);
ASSERT_OK(GenerateAndAddExternalFile(
options, {11, 15, 19}, ValueType::kTypeDeletion, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {11, 15, 19},
ValueType::kTypeDeletion, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 2);
ASSERT_OK(GenerateAndAddExternalFile(
options, {120, 130}, ValueType::kTypeMerge, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {120, 130},
ValueType::kTypeMerge, file_id++,
write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 2);
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 130}, ValueType::kTypeDeletion, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 130},
ValueType::kTypeDeletion, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 3);
ASSERT_OK(GenerateAndAddExternalFile(
options, {120}, {ValueType::kTypeValue}, {{120, 135}}, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 4);
ASSERT_OK(GenerateAndAddExternalFile(
options, {}, {}, {{110, 120}}, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {}, {}, {{110, 120}},
file_id++, write_global_seqno,
&true_data));
// The range deletion ends on a key, but it doesn't actually delete
// this key because the largest key in the range is exclusive. Still,
// it counts as an overlap so a new seqno will be assigned.
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 5);
ASSERT_OK(GenerateAndAddExternalFile(
options, {}, {}, {{100, 109}}, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {}, {}, {{100, 109}},
file_id++, write_global_seqno,
&true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 5);
@@ -433,21 +426,21 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMultipleValueType) {
}
SequenceNumber last_seqno = dbfull()->GetLatestSequenceNumber();
ASSERT_OK(GenerateAndAddExternalFile(
options, {60, 61, 62}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {60, 61, 62},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno);
ASSERT_OK(GenerateAndAddExternalFile(
options, {40, 41, 42}, ValueType::kTypeMerge, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {40, 41, 42},
ValueType::kTypeMerge, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 1);
ASSERT_OK(GenerateAndAddExternalFile(
options, {20, 30, 40}, ValueType::kTypeDeletion, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {20, 30, 40},
ValueType::kTypeDeletion, file_id++,
write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 2);
@@ -455,29 +448,29 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMultipleValueType) {
// We will need a seqno for the file regardless if the file overwrite
// keys in the DB or not because we have a snapshot
ASSERT_OK(GenerateAndAddExternalFile(
options, {1000, 1002}, ValueType::kTypeMerge, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1000, 1002},
ValueType::kTypeMerge, file_id++,
write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 3);
ASSERT_OK(GenerateAndAddExternalFile(
options, {2000, 3002}, ValueType::kTypeMerge, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {2000, 3002},
ValueType::kTypeMerge, file_id++,
write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 4);
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 20, 40, 100, 150}, ValueType::kTypeMerge, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 20, 40, 100, 150},
ValueType::kTypeMerge, file_id++,
write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 5);
db_->ReleaseSnapshot(snapshot);
ASSERT_OK(GenerateAndAddExternalFile(
options, {5000, 5001}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {5000, 5001},
ValueType::kTypeValue, file_id++,
write_global_seqno, &true_data));
// No snapshot anymore, no need to assign a seqno
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 5);
@@ -487,8 +480,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMultipleValueType) {
}
TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
bool write_global_seqno = GetParam();
do {
Options options = CurrentOptions();
options.merge_operator.reset(new TestPutOperator());
@@ -501,8 +493,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {1, 2, 3, 4, 5, 6},
{ValueType::kTypeValue, ValueType::kTypeMerge, ValueType::kTypeValue,
ValueType::kTypeMerge, ValueType::kTypeValue, ValueType::kTypeMerge},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 0);
@@ -510,8 +501,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {10, 11, 12, 13},
{ValueType::kTypeValue, ValueType::kTypeMerge, ValueType::kTypeValue,
ValueType::kTypeMerge},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 0);
@@ -519,8 +509,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {1, 4, 6},
{ValueType::kTypeDeletion, ValueType::kTypeValue,
ValueType::kTypeMerge},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 1);
@@ -528,22 +517,19 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {11, 15, 19},
{ValueType::kTypeDeletion, ValueType::kTypeMerge,
ValueType::kTypeValue},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 2);
ASSERT_OK(GenerateAndAddExternalFile(
options, {120, 130}, {ValueType::kTypeValue, ValueType::kTypeMerge},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 2);
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 130}, {ValueType::kTypeMerge, ValueType::kTypeDeletion},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 3);
@@ -551,16 +537,14 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {150, 151, 152},
{ValueType::kTypeValue, ValueType::kTypeMerge,
ValueType::kTypeDeletion},
{{150, 160}, {180, 190}}, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
{{150, 160}, {180, 190}}, file_id++, write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 3);
ASSERT_OK(GenerateAndAddExternalFile(
options, {150, 151, 152},
{ValueType::kTypeValue, ValueType::kTypeMerge, ValueType::kTypeValue},
{{200, 250}}, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
{{200, 250}}, file_id++, write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 4);
@@ -568,8 +552,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {300, 301, 302},
{ValueType::kTypeValue, ValueType::kTypeMerge,
ValueType::kTypeDeletion},
{{1, 2}, {152, 154}}, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
{{1, 2}, {152, 154}}, file_id++, write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 5);
@@ -583,8 +566,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
ASSERT_OK(GenerateAndAddExternalFile(
options, {60, 61, 62},
{ValueType::kTypeValue, ValueType::kTypeMerge, ValueType::kTypeValue},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File doesn't overwrite any keys, no seqno needed
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno);
@@ -592,8 +574,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {40, 41, 42},
{ValueType::kTypeValue, ValueType::kTypeDeletion,
ValueType::kTypeDeletion},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 1);
@@ -601,8 +582,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {20, 30, 40},
{ValueType::kTypeDeletion, ValueType::kTypeDeletion,
ValueType::kTypeDeletion},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// File overwrites some keys, a seqno will be assigned
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 2);
@@ -612,15 +592,13 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
// keys in the DB or not because we have a snapshot
ASSERT_OK(GenerateAndAddExternalFile(
options, {1000, 1002}, {ValueType::kTypeValue, ValueType::kTypeMerge},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 3);
ASSERT_OK(GenerateAndAddExternalFile(
options, {2000, 3002}, {ValueType::kTypeValue, ValueType::kTypeMerge},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 4);
@@ -628,8 +606,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
options, {1, 20, 40, 100, 150},
{ValueType::kTypeDeletion, ValueType::kTypeDeletion,
ValueType::kTypeValue, ValueType::kTypeMerge, ValueType::kTypeMerge},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// A global seqno will be assigned anyway because of the snapshot
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 5);
@@ -637,8 +614,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestFileWithMixedValueType) {
ASSERT_OK(GenerateAndAddExternalFile(
options, {5000, 5001}, {ValueType::kTypeValue, ValueType::kTypeMerge},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
// No snapshot anymore, no need to assign a seqno
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno + 5);
@@ -714,15 +690,13 @@ TEST_P(ExternalSSTFileBasicTest, IngestionWithRangeDeletions) {
ASSERT_EQ(0, NumTableFilesAtLevel(kNumLevels - 2));
ASSERT_EQ(1, NumTableFilesAtLevel(kNumLevels - 1));
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
bool write_global_seqno = GetParam();
// overlaps with L0 file but not memtable, so flush is skipped and file is
// ingested into L0
SequenceNumber last_seqno = dbfull()->GetLatestSequenceNumber();
ASSERT_OK(GenerateAndAddExternalFile(
options, {60, 90}, {ValueType::kTypeValue, ValueType::kTypeValue},
{{65, 70}, {70, 85}}, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
{{65, 70}, {70, 85}}, file_id++, write_global_seqno, &true_data));
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), ++last_seqno);
ASSERT_EQ(2, NumTableFilesAtLevel(0));
ASSERT_EQ(0, NumTableFilesAtLevel(kNumLevels - 2));
@@ -732,8 +706,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestionWithRangeDeletions) {
// file is ingested into L5
ASSERT_OK(GenerateAndAddExternalFile(
options, {10, 40}, {ValueType::kTypeValue, ValueType::kTypeValue},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), ++last_seqno);
ASSERT_EQ(2, NumTableFilesAtLevel(0));
ASSERT_EQ(1, NumTableFilesAtLevel(kNumLevels - 2));
@@ -741,9 +714,8 @@ TEST_P(ExternalSSTFileBasicTest, IngestionWithRangeDeletions) {
// overlaps with L5 file but not memtable or L0 file, so flush is skipped and
// file is ingested into L4
ASSERT_OK(GenerateAndAddExternalFile(
options, {}, {}, {{5, 15}}, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {}, {}, {{5, 15}}, file_id++,
write_global_seqno, &true_data));
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), ++last_seqno);
ASSERT_EQ(2, NumTableFilesAtLevel(0));
ASSERT_EQ(1, NumTableFilesAtLevel(kNumLevels - 2));
@@ -755,8 +727,7 @@ TEST_P(ExternalSSTFileBasicTest, IngestionWithRangeDeletions) {
// count increases by two.
ASSERT_OK(GenerateAndAddExternalFile(
options, {100, 140}, {ValueType::kTypeValue, ValueType::kTypeValue},
file_id++, write_global_seqno, verify_checksums_before_ingest,
&true_data));
file_id++, write_global_seqno, &true_data));
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), ++last_seqno);
ASSERT_EQ(4, NumTableFilesAtLevel(0));
ASSERT_EQ(1, NumTableFilesAtLevel(kNumLevels - 2));
@@ -769,101 +740,15 @@ TEST_P(ExternalSSTFileBasicTest, IngestionWithRangeDeletions) {
// seqnum.
ASSERT_OK(GenerateAndAddExternalFile(
options, {151, 175}, {ValueType::kTypeValue, ValueType::kTypeValue},
{{160, 200}}, file_id++, write_global_seqno,
verify_checksums_before_ingest, &true_data));
{{160, 200}}, file_id++, write_global_seqno, &true_data));
ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), last_seqno);
ASSERT_EQ(4, NumTableFilesAtLevel(0));
ASSERT_EQ(1, NumTableFilesAtLevel(kNumLevels - 2));
ASSERT_EQ(2, NumTableFilesAtLevel(options.num_levels - 1));
}
TEST_P(ExternalSSTFileBasicTest, IngestFileWithBadBlockChecksum) {
bool change_checksum_called = false;
const auto& change_checksum = [&](void* arg) {
if (!change_checksum_called) {
char* buf = reinterpret_cast<char*>(arg);
assert(nullptr != buf);
buf[0] ^= 0x1;
change_checksum_called = true;
}
};
SyncPoint::GetInstance()->DisableProcessing();
SyncPoint::GetInstance()->ClearAllCallBacks();
SyncPoint::GetInstance()->SetCallBack(
"BlockBasedTableBuilder::WriteRawBlock:TamperWithChecksum",
change_checksum);
SyncPoint::GetInstance()->EnableProcessing();
int file_id = 0;
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
do {
Options options = CurrentOptions();
DestroyAndReopen(options);
std::map<std::string, std::string> true_data;
Status s = GenerateAndAddExternalFile(
options, {1, 2, 3, 4, 5, 6}, ValueType::kTypeValue, file_id++,
write_global_seqno, verify_checksums_before_ingest, &true_data);
if (verify_checksums_before_ingest) {
ASSERT_NOK(s);
} else {
ASSERT_OK(s);
}
change_checksum_called = false;
} while (ChangeOptionsForFileIngestionTest());
}
TEST_P(ExternalSSTFileBasicTest, IngestFileWithFirstByteTampered) {
SyncPoint::GetInstance()->DisableProcessing();
int file_id = 0;
EnvOptions env_options;
do {
Options options = CurrentOptions();
std::string file_path = sst_files_dir_ + ToString(file_id++);
SstFileWriter sst_file_writer(env_options, options);
Status s = sst_file_writer.Open(file_path);
ASSERT_OK(s);
for (int i = 0; i != 100; ++i) {
std::string key = Key(i);
std::string value = Key(i) + ToString(0);
ASSERT_OK(sst_file_writer.Put(key, value));
}
ASSERT_OK(sst_file_writer.Finish());
{
// Get file size
uint64_t file_size = 0;
ASSERT_OK(env_->GetFileSize(file_path, &file_size));
ASSERT_GT(file_size, 8);
std::unique_ptr<RandomRWFile> rwfile;
ASSERT_OK(env_->NewRandomRWFile(file_path, &rwfile, EnvOptions()));
// Manually corrupt the file
// We deterministically corrupt the first byte because we currently
// cannot choose a random offset. The reason for this limitation is that
// we do not checksum property block at present.
const uint64_t offset = 0;
char scratch[8] = {0};
Slice buf;
ASSERT_OK(rwfile->Read(offset, sizeof(scratch), &buf, scratch));
scratch[0] ^= 0xff; // flip one bit
ASSERT_OK(rwfile->Write(offset, buf));
}
// Ingest file.
IngestExternalFileOptions ifo;
ifo.write_global_seqno = std::get<0>(GetParam());
ifo.verify_checksums_before_ingest = std::get<1>(GetParam());
s = db_->IngestExternalFile({file_path}, ifo);
if (ifo.verify_checksums_before_ingest) {
ASSERT_NOK(s);
} else {
ASSERT_OK(s);
}
} while (ChangeOptionsForFileIngestionTest());
}
INSTANTIATE_TEST_CASE_P(ExternalSSTFileBasicTest, ExternalSSTFileBasicTest,
testing::Values(std::make_tuple(true, true),
std::make_tuple(true, false),
std::make_tuple(false, true),
std::make_tuple(false, false)));
testing::Bool());
#endif // ROCKSDB_LITE
-7
View File
@@ -316,13 +316,6 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
return status;
}
if (ingestion_options_.verify_checksums_before_ingest) {
status = table_reader->VerifyChecksum();
}
if (!status.ok()) {
return status;
}
// Get the external file properties
auto props = table_reader->GetTableProperties();
const auto& uprops = props->user_collected_properties;
+64 -86
View File
@@ -15,9 +15,8 @@
namespace rocksdb {
class ExternalSSTFileTest
: public DBTestBase,
public ::testing::WithParamInterface<std::tuple<bool, bool>> {
class ExternalSSTFileTest : public DBTestBase,
public ::testing::WithParamInterface<bool> {
public:
ExternalSSTFileTest() : DBTestBase("/external_sst_file_test") {
sst_files_dir_ = dbname_ + "/sst_files/";
@@ -33,7 +32,7 @@ class ExternalSSTFileTest
const Options options,
std::vector<std::pair<std::string, std::string>> data, int file_id = -1,
bool allow_global_seqno = false, bool write_global_seqno = false,
bool verify_checksums_before_ingest = true, bool sort_data = false,
bool sort_data = false,
std::map<std::string, std::string>* true_data = nullptr,
ColumnFamilyHandle* cfh = nullptr) {
// Generate a file id if not provided
@@ -77,7 +76,6 @@ class ExternalSSTFileTest
IngestExternalFileOptions ifo;
ifo.allow_global_seqno = allow_global_seqno;
ifo.write_global_seqno = allow_global_seqno ? write_global_seqno : false;
ifo.verify_checksums_before_ingest = verify_checksums_before_ingest;
if (cfh) {
s = db_->IngestExternalFile(cfh, {file_path}, ifo);
} else {
@@ -157,32 +155,31 @@ class ExternalSSTFileTest
Status GenerateAndAddExternalFile(
const Options options, std::vector<std::pair<int, std::string>> data,
int file_id = -1, bool allow_global_seqno = false,
bool write_global_seqno = false,
bool verify_checksums_before_ingest = true, bool sort_data = false,
bool write_global_seqno = false, bool sort_data = false,
std::map<std::string, std::string>* true_data = nullptr,
ColumnFamilyHandle* cfh = nullptr) {
std::vector<std::pair<std::string, std::string>> file_data;
for (auto& entry : data) {
file_data.emplace_back(Key(entry.first), entry.second);
}
return GenerateAndAddExternalFile(
options, file_data, file_id, allow_global_seqno, write_global_seqno,
verify_checksums_before_ingest, sort_data, true_data, cfh);
return GenerateAndAddExternalFile(options, file_data, file_id,
allow_global_seqno, write_global_seqno,
sort_data, true_data, cfh);
}
Status GenerateAndAddExternalFile(
const Options options, std::vector<int> keys, int file_id = -1,
bool allow_global_seqno = false, bool write_global_seqno = false,
bool verify_checksums_before_ingest = true, bool sort_data = false,
bool sort_data = false,
std::map<std::string, std::string>* true_data = nullptr,
ColumnFamilyHandle* cfh = nullptr) {
std::vector<std::pair<std::string, std::string>> file_data;
for (auto& k : keys) {
file_data.emplace_back(Key(k), Key(k) + ToString(file_id));
}
return GenerateAndAddExternalFile(
options, file_data, file_id, allow_global_seqno, write_global_seqno,
verify_checksums_before_ingest, sort_data, true_data, cfh);
return GenerateAndAddExternalFile(options, file_data, file_id,
allow_global_seqno, write_global_seqno,
sort_data, true_data, cfh);
}
Status DeprecatedAddFile(const std::vector<std::string>& files,
@@ -1160,13 +1157,13 @@ TEST_P(ExternalSSTFileTest, PickedLevel) {
std::map<std::string, std::string> true_data;
// File 0 will go to last level (L3)
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 10}, -1, false, false, true,
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 10}, -1, false, false,
false, &true_data));
EXPECT_EQ(FilesPerLevel(), "0,0,0,1");
// File 1 will go to level L2 (since it overlap with file 0 in L3)
ASSERT_OK(GenerateAndAddExternalFile(options, {2, 9}, -1, false, false, true,
false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {2, 9}, -1, false, false, false,
&true_data));
EXPECT_EQ(FilesPerLevel(), "0,0,1,1");
rocksdb::SyncPoint::GetInstance()->LoadDependency({
@@ -1195,13 +1192,13 @@ TEST_P(ExternalSSTFileTest, PickedLevel) {
// This file overlaps with file 0 (L3), file 1 (L2) and the
// output of compaction going to L1
ASSERT_OK(GenerateAndAddExternalFile(options, {4, 7}, -1, false, false, true,
false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {4, 7}, -1, false, false, false,
&true_data));
EXPECT_EQ(FilesPerLevel(), "5,0,1,1");
// This file does not overlap with any file or with the running compaction
ASSERT_OK(GenerateAndAddExternalFile(options, {9000, 9001}, -1, false, false,
true, false, &true_data));
false, &true_data));
EXPECT_EQ(FilesPerLevel(), "5,0,1,2");
// Hold compaction from finishing
@@ -1321,7 +1318,7 @@ TEST_F(ExternalSSTFileTest, IngestNonExistingFile) {
ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
ASSERT_OK(dbfull()->TEST_WaitForCompact(true));
// After full compaction, there should be only 1 file.
std::vector<std::string> files;
env_->GetChildren(dbname_, &files);
@@ -1432,12 +1429,12 @@ TEST_F(ExternalSSTFileTest, PickedLevelDynamic) {
// This file overlaps with the output of the compaction (going to L3)
// so the file will be added to L0 since L3 is the base level
ASSERT_OK(GenerateAndAddExternalFile(options, {31, 32, 33, 34}, -1, false,
false, true, false, &true_data));
false, false, &true_data));
EXPECT_EQ(FilesPerLevel(), "5");
// This file does not overlap with the current running compactiong
ASSERT_OK(GenerateAndAddExternalFile(options, {9000, 9001}, -1, false, false,
true, false, &true_data));
false, &true_data));
EXPECT_EQ(FilesPerLevel(), "5,0,0,1");
// Hold compaction from finishing
@@ -1452,25 +1449,25 @@ TEST_F(ExternalSSTFileTest, PickedLevelDynamic) {
Reopen(options);
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 15, 19}, -1, false, false,
true, false, &true_data));
false, &true_data));
ASSERT_EQ(FilesPerLevel(), "1,0,0,3");
ASSERT_OK(GenerateAndAddExternalFile(options, {1000, 1001, 1002}, -1, false,
false, true, false, &true_data));
false, false, &true_data));
ASSERT_EQ(FilesPerLevel(), "1,0,0,4");
ASSERT_OK(GenerateAndAddExternalFile(options, {500, 600, 700}, -1, false,
false, true, false, &true_data));
false, false, &true_data));
ASSERT_EQ(FilesPerLevel(), "1,0,0,5");
// File 5 overlaps with file 2 (L3 / base level)
ASSERT_OK(GenerateAndAddExternalFile(options, {2, 10}, -1, false, false, true,
ASSERT_OK(GenerateAndAddExternalFile(options, {2, 10}, -1, false, false,
false, &true_data));
ASSERT_EQ(FilesPerLevel(), "2,0,0,5");
// File 6 overlaps with file 2 (L3 / base level) and file 5 (L0)
ASSERT_OK(GenerateAndAddExternalFile(options, {3, 9}, -1, false, false, true,
false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {3, 9}, -1, false, false, false,
&true_data));
ASSERT_EQ(FilesPerLevel(), "3,0,0,5");
// Verify data in files
@@ -1489,7 +1486,7 @@ TEST_F(ExternalSSTFileTest, PickedLevelDynamic) {
// File 7 overlaps with file 4 (L3)
ASSERT_OK(GenerateAndAddExternalFile(options, {650, 651, 652}, -1, false,
false, true, false, &true_data));
false, false, &true_data));
ASSERT_EQ(FilesPerLevel(), "5,0,0,5");
VerifyDBFromMap(true_data, &kcnt, false);
@@ -1629,8 +1626,7 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoRandomized) {
options.level0_slowdown_writes_trigger = 256;
options.level0_stop_writes_trigger = 256;
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
bool write_global_seqno = GetParam();
for (int iter = 0; iter < 2; iter++) {
bool write_to_memtable = (iter == 0);
DestroyAndReopen(options);
@@ -1654,9 +1650,9 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoRandomized) {
true_data[entry.first] = entry.second;
}
} else {
ASSERT_OK(GenerateAndAddExternalFile(
options, random_data, -1, true, write_global_seqno,
verify_checksums_before_ingest, true, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, random_data, -1, true,
write_global_seqno, true,
&true_data));
}
}
size_t kcnt = 0;
@@ -1685,11 +1681,9 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoAssignedLevel) {
for (int i = 0; i <= 20; i++) {
file_data.emplace_back(Key(i), "L4");
}
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
ASSERT_OK(GenerateAndAddExternalFile(
options, file_data, -1, true, write_global_seqno,
verify_checksums_before_ingest, false, &true_data));
bool write_global_seqno = GetParam();
ASSERT_OK(GenerateAndAddExternalFile(options, file_data, -1, true,
write_global_seqno, false, &true_data));
// This file dont overlap with anything in the DB, will go to L4
ASSERT_EQ("0,0,0,0,1", FilesPerLevel());
@@ -1699,9 +1693,8 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoAssignedLevel) {
for (int i = 80; i <= 130; i++) {
file_data.emplace_back(Key(i), "L0");
}
ASSERT_OK(GenerateAndAddExternalFile(
options, file_data, -1, true, write_global_seqno,
verify_checksums_before_ingest, false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, file_data, -1, true,
write_global_seqno, false, &true_data));
// This file overlap with the memtable, so it will flush it and add
// it self to L0
@@ -1712,9 +1705,8 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoAssignedLevel) {
for (int i = 30; i <= 50; i++) {
file_data.emplace_back(Key(i), "L4");
}
ASSERT_OK(GenerateAndAddExternalFile(
options, file_data, -1, true, write_global_seqno,
verify_checksums_before_ingest, false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, file_data, -1, true,
write_global_seqno, false, &true_data));
// This file dont overlap with anything in the DB and fit in L4 as well
ASSERT_EQ("2,0,0,0,2", FilesPerLevel());
@@ -1724,9 +1716,8 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoAssignedLevel) {
for (int i = 10; i <= 40; i++) {
file_data.emplace_back(Key(i), "L3");
}
ASSERT_OK(GenerateAndAddExternalFile(
options, file_data, -1, true, write_global_seqno,
verify_checksums_before_ingest, false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, file_data, -1, true,
write_global_seqno, false, &true_data));
// This file overlap with files in L4, we will ingest it in L3
ASSERT_EQ("2,0,0,1,2", FilesPerLevel());
@@ -1749,20 +1740,17 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoMemtableFlush) {
&entries_in_memtable);
ASSERT_GE(entries_in_memtable, 1);
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
bool write_global_seqno = GetParam();
// No need for flush
ASSERT_OK(GenerateAndAddExternalFile(
options, {90, 100, 110}, -1, true, write_global_seqno,
verify_checksums_before_ingest, false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {90, 100, 110}, -1, true,
write_global_seqno, false, &true_data));
db_->GetIntProperty(DB::Properties::kNumEntriesActiveMemTable,
&entries_in_memtable);
ASSERT_GE(entries_in_memtable, 1);
// This file will flush the memtable
ASSERT_OK(GenerateAndAddExternalFile(
options, {19, 20, 21}, -1, true, write_global_seqno,
verify_checksums_before_ingest, false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {19, 20, 21}, -1, true,
write_global_seqno, false, &true_data));
db_->GetIntProperty(DB::Properties::kNumEntriesActiveMemTable,
&entries_in_memtable);
ASSERT_EQ(entries_in_memtable, 0);
@@ -1776,17 +1764,15 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoMemtableFlush) {
ASSERT_GE(entries_in_memtable, 1);
// No need for flush, this file keys fit between the memtable keys
ASSERT_OK(GenerateAndAddExternalFile(
options, {202, 203, 204}, -1, true, write_global_seqno,
verify_checksums_before_ingest, false, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {202, 203, 204}, -1, true,
write_global_seqno, false, &true_data));
db_->GetIntProperty(DB::Properties::kNumEntriesActiveMemTable,
&entries_in_memtable);
ASSERT_GE(entries_in_memtable, 1);
// This file will flush the memtable
ASSERT_OK(GenerateAndAddExternalFile(
options, {206, 207}, -1, true, false, write_global_seqno,
verify_checksums_before_ingest, &true_data));
ASSERT_OK(GenerateAndAddExternalFile(options, {206, 207}, -1, true, false,
write_global_seqno, &true_data));
db_->GetIntProperty(DB::Properties::kNumEntriesActiveMemTable,
&entries_in_memtable);
ASSERT_EQ(entries_in_memtable, 0);
@@ -1804,16 +1790,13 @@ TEST_P(ExternalSSTFileTest, L0SortingIssue) {
ASSERT_OK(Put(Key(1), "memtable"));
ASSERT_OK(Put(Key(10), "memtable"));
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
bool write_global_seqno = GetParam();
// No Flush needed, No global seqno needed, Ingest in L1
ASSERT_OK(GenerateAndAddExternalFile(options, {7, 8}, -1, true,
write_global_seqno,
verify_checksums_before_ingest, false));
write_global_seqno, false));
// No Flush needed, but need a global seqno, Ingest in L0
ASSERT_OK(GenerateAndAddExternalFile(options, {7, 8}, -1, true,
write_global_seqno,
verify_checksums_before_ingest, false));
write_global_seqno, false));
printf("%s\n", FilesPerLevel().c_str());
// Overwrite what we added using external files
@@ -2049,12 +2032,11 @@ TEST_P(ExternalSSTFileTest, IngestionListener) {
options.listeners.emplace_back(listener);
CreateAndReopenWithCF({"koko", "toto"}, options);
bool write_global_seqno = std::get<0>(GetParam());
bool verify_checksums_before_ingest = std::get<1>(GetParam());
bool write_global_seqno = GetParam();
// Ingest into default cf
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 2}, -1, true, write_global_seqno,
verify_checksums_before_ingest, true, nullptr, handles_[0]));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 2}, -1, true,
write_global_seqno, true, nullptr,
handles_[0]));
ASSERT_EQ(listener->ingested_files.size(), 1);
ASSERT_EQ(listener->ingested_files.back().cf_name, "default");
ASSERT_EQ(listener->ingested_files.back().global_seqno, 0);
@@ -2064,9 +2046,9 @@ TEST_P(ExternalSSTFileTest, IngestionListener) {
"default");
// Ingest into cf1
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 2}, -1, true, write_global_seqno,
verify_checksums_before_ingest, true, nullptr, handles_[1]));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 2}, -1, true,
write_global_seqno, true, nullptr,
handles_[1]));
ASSERT_EQ(listener->ingested_files.size(), 2);
ASSERT_EQ(listener->ingested_files.back().cf_name, "koko");
ASSERT_EQ(listener->ingested_files.back().global_seqno, 0);
@@ -2076,9 +2058,9 @@ TEST_P(ExternalSSTFileTest, IngestionListener) {
"koko");
// Ingest into cf2
ASSERT_OK(GenerateAndAddExternalFile(
options, {1, 2}, -1, true, write_global_seqno,
verify_checksums_before_ingest, true, nullptr, handles_[2]));
ASSERT_OK(GenerateAndAddExternalFile(options, {1, 2}, -1, true,
write_global_seqno, true, nullptr,
handles_[2]));
ASSERT_EQ(listener->ingested_files.size(), 3);
ASSERT_EQ(listener->ingested_files.back().cf_name, "toto");
ASSERT_EQ(listener->ingested_files.back().global_seqno, 0);
@@ -2144,8 +2126,7 @@ TEST_P(ExternalSSTFileTest, IngestBehind) {
IngestExternalFileOptions ifo;
ifo.allow_global_seqno = true;
ifo.ingest_behind = true;
ifo.write_global_seqno = std::get<0>(GetParam());
ifo.verify_checksums_before_ingest = std::get<1>(GetParam());
ifo.write_global_seqno = GetParam();
// Can't ingest behind since allow_ingest_behind isn't set to true
ASSERT_NOK(GenerateAndAddExternalFileIngestBehind(options, ifo,
@@ -2234,10 +2215,7 @@ TEST_F(ExternalSSTFileTest, SkipBloomFilter) {
}
INSTANTIATE_TEST_CASE_P(ExternalSSTFileTest, ExternalSSTFileTest,
testing::Values(std::make_tuple(false, false),
std::make_tuple(false, true),
std::make_tuple(true, false),
std::make_tuple(true, true)));
testing::Bool());
} // namespace rocksdb
-10
View File
@@ -211,8 +211,6 @@ Status FlushJob::Run(LogsWithPrepTracker* prep_tracker,
uint64_t prev_fsync_nanos = 0;
uint64_t prev_range_sync_nanos = 0;
uint64_t prev_prepare_write_nanos = 0;
uint64_t prev_cpu_write_nanos = 0;
uint64_t prev_cpu_read_nanos = 0;
if (measure_io_stats_) {
prev_perf_level = GetPerfLevel();
SetPerfLevel(PerfLevel::kEnableTime);
@@ -220,8 +218,6 @@ Status FlushJob::Run(LogsWithPrepTracker* prep_tracker,
prev_fsync_nanos = IOSTATS(fsync_nanos);
prev_range_sync_nanos = IOSTATS(range_sync_nanos);
prev_prepare_write_nanos = IOSTATS(prepare_write_nanos);
prev_cpu_write_nanos = IOSTATS(cpu_write_nanos);
prev_cpu_read_nanos = IOSTATS(cpu_read_nanos);
}
// This will release and re-acquire the mutex.
@@ -273,10 +269,6 @@ Status FlushJob::Run(LogsWithPrepTracker* prep_tracker,
stream << "file_fsync_nanos" << (IOSTATS(fsync_nanos) - prev_fsync_nanos);
stream << "file_prepare_write_nanos"
<< (IOSTATS(prepare_write_nanos) - prev_prepare_write_nanos);
stream << "file_cpu_write_nanos"
<< (IOSTATS(cpu_write_nanos) - prev_cpu_write_nanos);
stream << "file_cpu_read_nanos"
<< (IOSTATS(cpu_read_nanos) - prev_cpu_read_nanos);
}
return s;
@@ -293,7 +285,6 @@ Status FlushJob::WriteLevel0Table() {
ThreadStatus::STAGE_FLUSH_WRITE_L0);
db_mutex_->AssertHeld();
const uint64_t start_micros = db_options_.env->NowMicros();
const uint64_t start_cpu_micros = db_options_.env->NowCPUNanos() / 1000;
Status s;
{
auto write_hint = cfd_->CalculateSSTWriteHint(0);
@@ -410,7 +401,6 @@ Status FlushJob::WriteLevel0Table() {
// Note that here we treat flush as level 0 compaction in internal stats
InternalStats::CompactionStats stats(CompactionReason::kFlush, 1);
stats.micros = db_options_.env->NowMicros() - start_micros;
stats.cpu_micros = db_options_.env->NowCPUNanos() / 1000 - start_cpu_micros;
stats.bytes_written = meta_.fd.GetFileSize();
MeasureTime(stats_, FLUSH_TIME, stats.micros);
cfd_->internal_stats()->AddCompactionStats(0 /* level */, stats);
+5 -10
View File
@@ -45,8 +45,6 @@ const std::map<LevelStatType, LevelStat> InternalStats::compaction_level_stats =
{LevelStatType::READ_MBPS, LevelStat{"ReadMBps", "Rd(MB/s)"}},
{LevelStatType::WRITE_MBPS, LevelStat{"WriteMBps", "Wr(MB/s)"}},
{LevelStatType::COMP_SEC, LevelStat{"CompSec", "Comp(sec)"}},
{LevelStatType::COMP_CPU_SEC,
LevelStat{"CompMergeCPU", "CompMergeCPU(sec)"}},
{LevelStatType::COMP_COUNT, LevelStat{"CompCount", "Comp(cnt)"}},
{LevelStatType::AVG_SEC, LevelStat{"AvgSec", "Avg(sec)"}},
{LevelStatType::KEY_IN, LevelStat{"KeyIn", "KeyIn"}},
@@ -66,7 +64,7 @@ void PrintLevelStatsHeader(char* buf, size_t len, const std::string& cf_name) {
};
int line_size = snprintf(
buf + written_size, len - written_size,
"Level %s %s %s %s %s %s %s %s %s %s %s %s %s %s %s %s %s %s\n",
"Level %s %s %s %s %s %s %s %s %s %s %s %s %s %s %s %s %s\n",
// Note that we skip COMPACTED_FILES and merge it with Files column
hdr(LevelStatType::NUM_FILES), hdr(LevelStatType::SIZE_BYTES),
hdr(LevelStatType::SCORE), hdr(LevelStatType::READ_GB),
@@ -74,9 +72,9 @@ void PrintLevelStatsHeader(char* buf, size_t len, const std::string& cf_name) {
hdr(LevelStatType::WRITE_GB), hdr(LevelStatType::W_NEW_GB),
hdr(LevelStatType::MOVED_GB), hdr(LevelStatType::WRITE_AMP),
hdr(LevelStatType::READ_MBPS), hdr(LevelStatType::WRITE_MBPS),
hdr(LevelStatType::COMP_SEC), hdr(LevelStatType::COMP_CPU_SEC),
hdr(LevelStatType::COMP_COUNT), hdr(LevelStatType::AVG_SEC),
hdr(LevelStatType::KEY_IN), hdr(LevelStatType::KEY_DROP));
hdr(LevelStatType::COMP_SEC), hdr(LevelStatType::COMP_COUNT),
hdr(LevelStatType::AVG_SEC), hdr(LevelStatType::KEY_IN),
hdr(LevelStatType::KEY_DROP));
written_size += line_size;
snprintf(buf + written_size, len - written_size, "%s\n",
@@ -108,7 +106,6 @@ void PrepareLevelStats(std::map<LevelStatType, double>* level_stats,
(*level_stats)[LevelStatType::WRITE_MBPS] =
stats.bytes_written / kMB / elapsed;
(*level_stats)[LevelStatType::COMP_SEC] = stats.micros / kMicrosInSec;
(*level_stats)[LevelStatType::COMP_CPU_SEC] = stats.cpu_micros / kMicrosInSec;
(*level_stats)[LevelStatType::COMP_COUNT] = stats.count;
(*level_stats)[LevelStatType::AVG_SEC] =
stats.count == 0 ? 0 : stats.micros / kMicrosInSec / stats.count;
@@ -135,8 +132,7 @@ void PrintLevelStats(char* buf, size_t len, const std::string& name,
"%5.1f " /* W-Amp */
"%8.1f " /* Rd(MB/s) */
"%8.1f " /* Wr(MB/s) */
"%9.2f " /* Comp(sec) */
"%17.2f " /* CompMergeCPU(sec) */
"%9.0f " /* Comp(sec) */
"%9d " /* Comp(cnt) */
"%8.3f " /* Avg(sec) */
"%7s " /* KeyIn */
@@ -157,7 +153,6 @@ void PrintLevelStats(char* buf, size_t len, const std::string& name,
stat_value.at(LevelStatType::READ_MBPS),
stat_value.at(LevelStatType::WRITE_MBPS),
stat_value.at(LevelStatType::COMP_SEC),
stat_value.at(LevelStatType::COMP_CPU_SEC),
static_cast<int>(stat_value.at(LevelStatType::COMP_COUNT)),
stat_value.at(LevelStatType::AVG_SEC),
NumberToHumanString(
+2 -10
View File
@@ -71,7 +71,6 @@ enum class LevelStatType {
READ_MBPS,
WRITE_MBPS,
COMP_SEC,
COMP_CPU_SEC,
COMP_COUNT,
AVG_SEC,
KEY_IN,
@@ -136,7 +135,6 @@ class InternalStats {
// compactions that produced data for the specified "level".
struct CompactionStats {
uint64_t micros;
uint64_t cpu_micros;
// The number of bytes read from all non-output levels
uint64_t bytes_read_non_output_levels;
@@ -174,7 +172,6 @@ class InternalStats {
explicit CompactionStats()
: micros(0),
cpu_micros(0),
bytes_read_non_output_levels(0),
bytes_read_output_level(0),
bytes_written(0),
@@ -193,7 +190,6 @@ class InternalStats {
explicit CompactionStats(CompactionReason reason, int c)
: micros(0),
cpu_micros(0),
bytes_read_non_output_levels(0),
bytes_read_output_level(0),
bytes_written(0),
@@ -218,14 +214,14 @@ class InternalStats {
explicit CompactionStats(const CompactionStats& c)
: micros(c.micros),
cpu_micros(c.cpu_micros),
bytes_read_non_output_levels(c.bytes_read_non_output_levels),
bytes_read_output_level(c.bytes_read_output_level),
bytes_written(c.bytes_written),
bytes_moved(c.bytes_moved),
num_input_files_in_non_output_levels(
c.num_input_files_in_non_output_levels),
num_input_files_in_output_level(c.num_input_files_in_output_level),
num_input_files_in_output_level(
c.num_input_files_in_output_level),
num_output_files(c.num_output_files),
num_input_records(c.num_input_records),
num_dropped_records(c.num_dropped_records),
@@ -238,7 +234,6 @@ class InternalStats {
void Clear() {
this->micros = 0;
this->cpu_micros = 0;
this->bytes_read_non_output_levels = 0;
this->bytes_read_output_level = 0;
this->bytes_written = 0;
@@ -257,7 +252,6 @@ class InternalStats {
void Add(const CompactionStats& c) {
this->micros += c.micros;
this->cpu_micros += c.cpu_micros;
this->bytes_read_non_output_levels += c.bytes_read_non_output_levels;
this->bytes_read_output_level += c.bytes_read_output_level;
this->bytes_written += c.bytes_written;
@@ -278,7 +272,6 @@ class InternalStats {
void Subtract(const CompactionStats& c) {
this->micros -= c.micros;
this->cpu_micros -= c.cpu_micros;
this->bytes_read_non_output_levels -= c.bytes_read_non_output_levels;
this->bytes_read_output_level -= c.bytes_read_output_level;
this->bytes_written -= c.bytes_written;
@@ -602,7 +595,6 @@ class InternalStats {
struct CompactionStats {
uint64_t micros;
uint64_t cpu_micros;
uint64_t bytes_read_non_output_levels;
uint64_t bytes_read_output_level;
uint64_t bytes_written;
+2 -2
View File
@@ -298,8 +298,8 @@ Status MergeHelper::MergeUntil(InternalIterator* iter,
// to combine the keys. Since VersionSet::SetupOtherInputs() always makes
// sure that all merge-operands on the same level get compacted together,
// this will simply lead to these merge operands moving to the next level.
bool surely_seen_the_beginning =
(hit_the_next_user_key || !iter->Valid()) && at_bottom;
bool surely_seen_the_beginning = (hit_the_next_user_key || !iter->Valid())
&& at_bottom;
if (surely_seen_the_beginning) {
// do a final merge with nullptr as the existing value and say
// bye to the merge type (it's now converted to a Put)
+65 -13
View File
@@ -134,6 +134,7 @@ class PlainTableDBTest : public testing::Test,
plain_table_options.huge_page_tlb_size = 0;
plain_table_options.encoding_type = kPrefix;
plain_table_options.full_scan_mode = false;
plain_table_options.store_index_in_file = false;
options.table_factory.reset(NewPlainTableFactory(plain_table_options));
options.memtable_factory.reset(NewHashLinkListRepFactory(4, 0, 3, true));
@@ -271,7 +272,8 @@ class TestPlainTableReader : public PlainTableReader {
std::unique_ptr<RandomAccessFileReader>&& file,
const ImmutableCFOptions& ioptions,
const SliceTransform* prefix_extractor,
bool* expect_bloom_not_match, uint32_t column_family_id,
bool* expect_bloom_not_match, bool store_index_in_file,
uint32_t column_family_id,
const std::string& column_family_name)
: PlainTableReader(ioptions, std::move(file), env_options, icomparator,
encoding_type, file_size, table_properties,
@@ -288,6 +290,17 @@ class TestPlainTableReader : public PlainTableReader {
TableProperties* props = const_cast<TableProperties*>(table_properties);
EXPECT_EQ(column_family_id, static_cast<uint32_t>(props->column_family_id));
EXPECT_EQ(column_family_name, props->column_family_name);
if (store_index_in_file) {
auto bloom_version_ptr = props->user_collected_properties.find(
PlainTablePropertyNames::kBloomVersion);
EXPECT_TRUE(bloom_version_ptr != props->user_collected_properties.end());
EXPECT_EQ(bloom_version_ptr->second, std::string("1"));
if (ioptions.bloom_locality > 0) {
auto num_blocks_ptr = props->user_collected_properties.find(
PlainTablePropertyNames::kNumBloomBlocks);
EXPECT_TRUE(num_blocks_ptr != props->user_collected_properties.end());
}
}
}
virtual ~TestPlainTableReader() {}
@@ -316,6 +329,7 @@ class TestPlainTableFactory : public PlainTableFactory {
bloom_bits_per_key_(options.bloom_bits_per_key),
hash_table_ratio_(options.hash_table_ratio),
index_sparseness_(options.index_sparseness),
store_index_in_file_(options.store_index_in_file),
expect_bloom_not_match_(expect_bloom_not_match),
column_family_id_(column_family_id),
column_family_name_(std::move(column_family_name)) {}
@@ -332,6 +346,22 @@ class TestPlainTableFactory : public PlainTableFactory {
true /* compression_type_missing */);
EXPECT_TRUE(s.ok());
if (store_index_in_file_) {
BlockHandle bloom_block_handle;
s = FindMetaBlock(file.get(), file_size, kPlainTableMagicNumber,
table_reader_options.ioptions,
BloomBlockBuilder::kBloomBlock, &bloom_block_handle,
/* compression_type_missing */ true);
EXPECT_TRUE(s.ok());
BlockHandle index_block_handle;
s = FindMetaBlock(file.get(), file_size, kPlainTableMagicNumber,
table_reader_options.ioptions,
PlainTableIndexBuilder::kPlainTableIndexBlock,
&index_block_handle, /* compression_type_missing */ true);
EXPECT_TRUE(s.ok());
}
auto& user_props = props->user_collected_properties;
auto encoding_type_prop =
user_props.find(PlainTablePropertyNames::kEncodingType);
@@ -345,7 +375,7 @@ class TestPlainTableFactory : public PlainTableFactory {
bloom_bits_per_key_, hash_table_ratio_, index_sparseness_, props,
std::move(file), table_reader_options.ioptions,
table_reader_options.prefix_extractor, expect_bloom_not_match_,
column_family_id_, column_family_name_));
store_index_in_file_, column_family_id_, column_family_name_));
*table = std::move(new_reader);
return s;
@@ -355,6 +385,7 @@ class TestPlainTableFactory : public PlainTableFactory {
int bloom_bits_per_key_;
double hash_table_ratio_;
size_t index_sparseness_;
bool store_index_in_file_;
bool* expect_bloom_not_match_;
const uint32_t column_family_id_;
const std::string column_family_name_;
@@ -364,8 +395,10 @@ TEST_P(PlainTableDBTest, Flush) {
for (size_t huge_page_tlb_size = 0; huge_page_tlb_size <= 2 * 1024 * 1024;
huge_page_tlb_size += 2 * 1024 * 1024) {
for (EncodingType encoding_type : {kPlain, kPrefix}) {
for (int bloom_bits = 0; bloom_bits <= 117; bloom_bits += 117) {
for (int total_order = 0; total_order <= 1; total_order++) {
for (int bloom_bits = 0; bloom_bits <= 117; bloom_bits += 117) {
for (int total_order = 0; total_order <= 1; total_order++) {
for (int store_index_in_file = 0; store_index_in_file <= 1;
++store_index_in_file) {
Options options = CurrentOptions();
options.create_if_missing = true;
// Set only one bucket to force bucket conflict.
@@ -381,6 +414,7 @@ TEST_P(PlainTableDBTest, Flush) {
plain_table_options.huge_page_tlb_size = huge_page_tlb_size;
plain_table_options.encoding_type = encoding_type;
plain_table_options.full_scan_mode = false;
plain_table_options.store_index_in_file = store_index_in_file;
options.table_factory.reset(
NewPlainTableFactory(plain_table_options));
@@ -393,6 +427,7 @@ TEST_P(PlainTableDBTest, Flush) {
plain_table_options.huge_page_tlb_size = huge_page_tlb_size;
plain_table_options.encoding_type = encoding_type;
plain_table_options.full_scan_mode = false;
plain_table_options.store_index_in_file = store_index_in_file;
options.table_factory.reset(
NewPlainTableFactory(plain_table_options));
@@ -418,15 +453,22 @@ TEST_P(PlainTableDBTest, Flush) {
auto row = ptc.begin();
auto tp = row->second;
ASSERT_EQ(total_order ? "4" : "12",
(tp->user_collected_properties)
.at("plain_table_hash_table_size"));
ASSERT_EQ(
"0",
(tp->user_collected_properties).at("plain_table_sub_index_size"));
if (!store_index_in_file) {
ASSERT_EQ(total_order ? "4" : "12",
(tp->user_collected_properties)
.at("plain_table_hash_table_size"));
ASSERT_EQ("0", (tp->user_collected_properties)
.at("plain_table_sub_index_size"));
} else {
ASSERT_EQ("0", (tp->user_collected_properties)
.at("plain_table_hash_table_size"));
ASSERT_EQ("0", (tp->user_collected_properties)
.at("plain_table_sub_index_size"));
}
ASSERT_EQ("v3", Get("1000000000000foo"));
ASSERT_EQ("v2", Get("0000000000000bar"));
}
}
}
}
}
@@ -436,11 +478,19 @@ TEST_P(PlainTableDBTest, Flush2) {
for (size_t huge_page_tlb_size = 0; huge_page_tlb_size <= 2 * 1024 * 1024;
huge_page_tlb_size += 2 * 1024 * 1024) {
for (EncodingType encoding_type : {kPlain, kPrefix}) {
for (int bloom_bits = 0; bloom_bits <= 117; bloom_bits += 117) {
for (int total_order = 0; total_order <= 1; total_order++) {
for (int bloom_bits = 0; bloom_bits <= 117; bloom_bits += 117) {
for (int total_order = 0; total_order <= 1; total_order++) {
for (int store_index_in_file = 0; store_index_in_file <= 1;
++store_index_in_file) {
if (encoding_type == kPrefix && total_order) {
continue;
}
if (!bloom_bits && store_index_in_file) {
continue;
}
if (total_order && store_index_in_file) {
continue;
}
bool expect_bloom_not_match = false;
Options options = CurrentOptions();
options.create_if_missing = true;
@@ -459,6 +509,7 @@ TEST_P(PlainTableDBTest, Flush2) {
plain_table_options.bloom_bits_per_key = bloom_bits;
plain_table_options.huge_page_tlb_size = huge_page_tlb_size;
plain_table_options.encoding_type = encoding_type;
plain_table_options.store_index_in_file = store_index_in_file;
options.table_factory.reset(new TestPlainTableFactory(
&expect_bloom_not_match, plain_table_options,
0 /* column_family_id */, kDefaultColumnFamilyName));
@@ -497,8 +548,9 @@ TEST_P(PlainTableDBTest, Flush2) {
}
expect_bloom_not_match = false;
}
}
}
}
}
}
}
}
+2 -2
View File
@@ -3044,8 +3044,8 @@ Status VersionSet::ProcessManifestWrites(
db_options_->manifest_preallocation_size);
std::unique_ptr<WritableFileWriter> file_writer(new WritableFileWriter(
std::move(descriptor_file), descriptor_fname, opt_env_opts, env_,
nullptr, db_options_->listeners));
std::move(descriptor_file), descriptor_fname, opt_env_opts, nullptr,
db_options_->listeners));
descriptor_log_.reset(
new log::Writer(std::move(file_writer), 0, false));
s = WriteSnapshot(descriptor_log_.get());
+1 -2
View File
@@ -844,8 +844,7 @@ class PosixEnv : public Env {
}
virtual uint64_t NowCPUNanos() override {
#if defined(OS_LINUX) || defined(OS_FREEBSD) || defined(OS_AIX) || \
defined(__MACH__)
#if defined(OS_LINUX) || defined(OS_FREEBSD) || defined(OS_AIX)
struct timespec ts;
clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts);
return static_cast<uint64_t>(ts.tv_sec) * 1000000000 + ts.tv_nsec;
-3
View File
@@ -18,9 +18,6 @@ struct CompactionJobStats {
// the elapsed time of this compaction in microseconds.
uint64_t elapsed_micros;
// the elapsed CPU time of this compaction in microseconds.
uint64_t cpu_micros;
// the number of compaction input records.
uint64_t num_input_records;
// the number of compaction input files.
-4
View File
@@ -44,10 +44,6 @@ struct IOStatsContext {
uint64_t prepare_write_nanos;
// time spent in Logger::Logv().
uint64_t logger_nanos;
// CPU time spent in write() and pwrite()
uint64_t cpu_write_nanos;
// CPU time spent in read() and pread()
uint64_t cpu_read_nanos;
};
// Get Thread-local IOStatsContext object pointer
+1 -6
View File
@@ -295,7 +295,7 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
std::vector<DbPath> cf_paths;
// Compaction concurrent thread limiter for the column family.
// If non-nullptr, use given concurrent thread limiter to control
// If non-nullptr, use given concurrent thread limiter to control
// the max outstanding compaction tasks. Limiter can be shared with
// multiple column families across db instances.
//
@@ -1344,11 +1344,6 @@ struct IngestExternalFileOptions {
// 2. Without writing external SST file, it's possible to do checksum.
// We have a plan to set this option to false by default in the future.
bool write_global_seqno = true;
// Set to true if you would like to verify the checksums of each block of the
// external SST file before ingestion.
// Warning: setting this to true causes slowdown in file ingestion because
// the external SST file has to be read.
bool verify_checksums_before_ingest = false;
};
// TraceOptions is used for StartTrace
-1
View File
@@ -352,7 +352,6 @@ enum Histograms : uint32_t {
DB_GET = 0,
DB_WRITE,
COMPACTION_TIME,
COMPACTION_CPU_TIME,
SUBCOMPACTION_SETUP_TIME,
TABLE_SYNC_MICROS,
COMPACTION_OUTFILE_SYNC_MICROS,
+1 -2
View File
@@ -351,11 +351,10 @@ struct PlainTableOptions {
// using the index.
bool full_scan_mode = false;
// THIS FEATURE IS REMOVED.
// @store_index_in_file: compute plain table index and bloom filter during
// file building and store it in file. When reading
// file, index will be mmaped instead of recomputation.
// bool store_index_in_file = false;
bool store_index_in_file = false;
};
// -- Plain Table with prefix-only seek
+1 -1
View File
@@ -69,7 +69,7 @@ class WriteBatchBase {
const SliceParts& key);
virtual Status SingleDelete(const SliceParts& key);
// If the database contains mappings in the range ["begin_key", "end_key"),
// If the database contains mappings in the range ["begin_key", "end_key"],
// erase them. Else do nothing.
virtual Status DeleteRange(ColumnFamilyHandle* column_family,
const Slice& begin_key, const Slice& end_key) = 0;
+3 -1
View File
@@ -20,7 +20,8 @@
jlong Java_org_rocksdb_PlainTableConfig_newTableFactoryHandle(
JNIEnv * /*env*/, jobject /*jobj*/, jint jkey_size,
jint jbloom_bits_per_key, jdouble jhash_table_ratio, jint jindex_sparseness,
jint jhuge_page_tlb_size, jbyte jencoding_type, jboolean jfull_scan_mode) {
jint jhuge_page_tlb_size, jbyte jencoding_type, jboolean jfull_scan_mode,
jboolean jstore_index_in_file) {
rocksdb::PlainTableOptions options = rocksdb::PlainTableOptions();
options.user_key_len = jkey_size;
options.bloom_bits_per_key = jbloom_bits_per_key;
@@ -29,6 +30,7 @@ jlong Java_org_rocksdb_PlainTableConfig_newTableFactoryHandle(
options.huge_page_tlb_size = jhuge_page_tlb_size;
options.encoding_type = static_cast<rocksdb::EncodingType>(jencoding_type);
options.full_scan_mode = jfull_scan_mode;
options.store_index_in_file = jstore_index_in_file;
return reinterpret_cast<jlong>(rocksdb::NewPlainTableFactory(options));
}
@@ -21,6 +21,8 @@ public class PlainTableConfig extends TableFormatConfig {
public static final EncodingType DEFAULT_ENCODING_TYPE =
EncodingType.kPlain;
public static final boolean DEFAULT_FULL_SCAN_MODE = false;
public static final boolean DEFAULT_STORE_INDEX_IN_FILE
= false;
public PlainTableConfig() {
keySize_ = VARIABLE_LENGTH;
@@ -30,6 +32,7 @@ public class PlainTableConfig extends TableFormatConfig {
hugePageTlbSize_ = DEFAULT_HUGE_TLB_SIZE;
encodingType_ = DEFAULT_ENCODING_TYPE;
fullScanMode_ = DEFAULT_FULL_SCAN_MODE;
storeIndexInFile_ = DEFAULT_STORE_INDEX_IN_FILE;
}
/**
@@ -208,10 +211,9 @@ public class PlainTableConfig extends TableFormatConfig {
* @param storeIndexInFile value indicating if index shall
* be stored in a file
* @return the reference to the current config.
* @deprecated
*/
@Deprecated
public PlainTableConfig setStoreIndexInFile(boolean storeIndexInFile) {
this.storeIndexInFile_ = storeIndexInFile;
return this;
}
@@ -220,20 +222,23 @@ public class PlainTableConfig extends TableFormatConfig {
* in a file.
*
* @return currently set value for store index in file.
* @deprecated
*/
@Deprecated
public boolean storeIndexInFile() {
return false;
return storeIndexInFile_;
}
@Override protected long newTableFactoryHandle() {
return newTableFactoryHandle(keySize_, bloomBitsPerKey_, hashTableRatio_, indexSparseness_,
hugePageTlbSize_, encodingType_.getValue(), fullScanMode_);
return newTableFactoryHandle(keySize_, bloomBitsPerKey_,
hashTableRatio_, indexSparseness_, hugePageTlbSize_,
encodingType_.getValue(), fullScanMode_,
storeIndexInFile_);
}
private native long newTableFactoryHandle(int keySize, int bloomBitsPerKey, double hashTableRatio,
int indexSparseness, int hugePageTlbSize, byte encodingType, boolean fullScanMode);
private native long newTableFactoryHandle(
int keySize, int bloomBitsPerKey,
double hashTableRatio, int indexSparseness,
int hugePageTlbSize, byte encodingType,
boolean fullScanMode, boolean storeIndexInFile);
private int keySize_;
private int bloomBitsPerKey_;
@@ -242,4 +247,5 @@ public class PlainTableConfig extends TableFormatConfig {
private int hugePageTlbSize_;
private EncodingType encodingType_;
private boolean fullScanMode_;
private boolean storeIndexInFile_;
}
@@ -70,6 +70,14 @@ public class PlainTableConfigTest {
plainTableConfig.setFullScanMode(true);
assertThat(plainTableConfig.fullScanMode()).isTrue(); }
@Test
public void storeIndexInFile() {
PlainTableConfig plainTableConfig = new PlainTableConfig();
plainTableConfig.setStoreIndexInFile(true);
assertThat(plainTableConfig.storeIndexInFile()).
isTrue();
}
@Test
public void plainTableConfig() {
try(final Options opt = new Options()) {
-7
View File
@@ -37,13 +37,6 @@ extern __thread IOStatsContext iostats_context;
PerfStepTimer iostats_step_timer_##metric(&(iostats_context.metric)); \
iostats_step_timer_##metric.Start();
// Declare and set start time of the timer
#define IOSTATS_CPU_TIMER_GUARD(metric, env) \
PerfStepTimer iostats_step_timer_##metric( \
&(iostats_context.metric), env, true, \
PerfLevel::kEnableTimeAndCPUTimeExceptForMutex); \
iostats_step_timer_##metric.Start();
#else // ROCKSDB_SUPPORT_THREAD_LOCAL
#define IOSTATS_ADD(metric, value)
-1
View File
@@ -184,7 +184,6 @@ const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {
{DB_GET, "rocksdb.db.get.micros"},
{DB_WRITE, "rocksdb.db.write.micros"},
{COMPACTION_TIME, "rocksdb.compaction.times.micros"},
{COMPACTION_CPU_TIME, "rocksdb.compaction.times.cpu_micros"},
{SUBCOMPACTION_SETUP_TIME, "rocksdb.subcompaction.setup.times.micros"},
{TABLE_SYNC_MICROS, "rocksdb.table.sync.micros"},
{COMPACTION_OUTFILE_SYNC_MICROS, "rocksdb.compaction.outfile.sync.micros"},
+1
View File
@@ -656,6 +656,7 @@ TEST_F(OptionsTest, GetPlainTableOptionsFromString) {
ASSERT_EQ(new_opt.huge_page_tlb_size, 4);
ASSERT_EQ(new_opt.encoding_type, EncodingType::kPrefix);
ASSERT_TRUE(new_opt.full_scan_mode);
ASSERT_TRUE(new_opt.store_index_in_file);
// unknown option
ASSERT_NOK(GetPlainTableOptionsFromString(table_opt,
-3
View File
@@ -634,9 +634,6 @@ void BlockBasedTableBuilder::WriteRawBlock(const Slice& block_contents,
}
assert(r->status.ok());
TEST_SYNC_POINT_CALLBACK(
"BlockBasedTableBuilder::WriteRawBlock:TamperWithChecksum",
static_cast<char*>(trailer));
r->status = r->file->Append(Slice(trailer, kBlockTrailerSize));
if (r->status.ok()) {
r->status = InsertBlockInCache(block_contents, type, handle);
+74 -1
View File
@@ -20,6 +20,7 @@
#include "table/plain_table_factory.h"
#include "db/dbformat.h"
#include "table/block_builder.h"
#include "table/bloom_block.h"
#include "table/plain_table_index.h"
#include "table/format.h"
#include "table/meta_blocks.h"
@@ -61,17 +62,34 @@ PlainTableBuilder::PlainTableBuilder(
int_tbl_prop_collector_factories,
uint32_t column_family_id, WritableFileWriter* file, uint32_t user_key_len,
EncodingType encoding_type, size_t index_sparseness,
const std::string& column_family_name)
uint32_t bloom_bits_per_key, const std::string& column_family_name,
uint32_t num_probes, size_t huge_page_tlb_size, double hash_table_ratio,
bool store_index_in_file)
: ioptions_(ioptions),
moptions_(moptions),
bloom_block_(num_probes),
file_(file),
bloom_bits_per_key_(bloom_bits_per_key),
huge_page_tlb_size_(huge_page_tlb_size),
encoder_(encoding_type, user_key_len, moptions.prefix_extractor.get(),
index_sparseness),
store_index_in_file_(store_index_in_file),
prefix_extractor_(moptions.prefix_extractor.get()) {
// Build index block and save it in the file if hash_table_ratio > 0
if (store_index_in_file_) {
assert(hash_table_ratio > 0 || IsTotalOrderMode());
index_builder_.reset(new PlainTableIndexBuilder(
&arena_, ioptions, moptions.prefix_extractor.get(), index_sparseness,
hash_table_ratio, huge_page_tlb_size_));
properties_.user_collected_properties
[PlainTablePropertyNames::kBloomVersion] = "1"; // For future use
}
properties_.fixed_key_len = user_key_len;
// for plain table, we put all the data in a big chuck.
properties_.num_data_blocks = 1;
// Fill it later if store_index_in_file_ == true
properties_.index_size = 0;
properties_.filter_size = 0;
// To support roll-back to previous version, now still use version 0 for
@@ -112,11 +130,26 @@ void PlainTableBuilder::Add(const Slice& key, const Slice& value) {
return;
}
// Store key hash
if (store_index_in_file_) {
if (moptions_.prefix_extractor == nullptr) {
keys_or_prefixes_hashes_.push_back(GetSliceHash(internal_key.user_key));
} else {
Slice prefix =
moptions_.prefix_extractor->Transform(internal_key.user_key);
keys_or_prefixes_hashes_.push_back(GetSliceHash(prefix));
}
}
// Write value
assert(offset_ <= std::numeric_limits<uint32_t>::max());
auto prev_offset = static_cast<uint32_t>(offset_);
// Write out the key
encoder_.AppendKey(key, file_, &offset_, meta_bytes_buf,
&meta_bytes_buf_size);
if (SaveIndexInFile()) {
index_builder_->AddKeyPrefix(GetPrefix(internal_key), prev_offset);
}
// Write value length
uint32_t value_size = static_cast<uint32_t>(value.size());
@@ -162,6 +195,46 @@ Status PlainTableBuilder::Finish() {
MetaIndexBuilder meta_index_builer;
if (store_index_in_file_ && (properties_.num_entries > 0)) {
assert(properties_.num_entries <= std::numeric_limits<uint32_t>::max());
Status s;
BlockHandle bloom_block_handle;
if (bloom_bits_per_key_ > 0) {
bloom_block_.SetTotalBits(
&arena_,
static_cast<uint32_t>(properties_.num_entries) * bloom_bits_per_key_,
ioptions_.bloom_locality, huge_page_tlb_size_, ioptions_.info_log);
PutVarint32(&properties_.user_collected_properties
[PlainTablePropertyNames::kNumBloomBlocks],
bloom_block_.GetNumBlocks());
bloom_block_.AddKeysHashes(keys_or_prefixes_hashes_);
Slice bloom_finish_result = bloom_block_.Finish();
properties_.filter_size = bloom_finish_result.size();
s = WriteBlock(bloom_finish_result, file_, &offset_, &bloom_block_handle);
if (!s.ok()) {
return s;
}
meta_index_builer.Add(BloomBlockBuilder::kBloomBlock, bloom_block_handle);
}
BlockHandle index_block_handle;
Slice index_finish_result = index_builder_->Finish();
properties_.index_size = index_finish_result.size();
s = WriteBlock(index_finish_result, file_, &offset_, &index_block_handle);
if (!s.ok()) {
return s;
}
meta_index_builer.Add(PlainTableIndexBuilder::kPlainTableIndexBlock,
index_block_handle);
}
// Calculate bloom block size and index block size
PropertyBlockBuilder property_block_builder;
// -- Add basic properties
+15 -1
View File
@@ -12,6 +12,8 @@
#include "rocksdb/status.h"
#include "rocksdb/table.h"
#include "rocksdb/table_properties.h"
#include "table/bloom_block.h"
#include "table/plain_table_index.h"
#include "table/plain_table_key_coding.h"
#include "table/table_builder.h"
@@ -35,7 +37,10 @@ class PlainTableBuilder: public TableBuilder {
int_tbl_prop_collector_factories,
uint32_t column_family_id, WritableFileWriter* file,
uint32_t user_key_size, EncodingType encoding_type,
size_t index_sparseness, const std::string& column_family_name);
size_t index_sparseness, uint32_t bloom_bits_per_key,
const std::string& column_family_name, uint32_t num_probes = 6,
size_t huge_page_tlb_size = 0, double hash_table_ratio = 0,
bool store_index_in_file = false);
// REQUIRES: Either Finish() or Abandon() has been called.
~PlainTableBuilder();
@@ -69,6 +74,8 @@ class PlainTableBuilder: public TableBuilder {
TableProperties GetTableProperties() const override { return properties_; }
bool SaveIndexInFile() const { return store_index_in_file_; }
private:
Arena arena_;
const ImmutableCFOptions& ioptions_;
@@ -76,12 +83,19 @@ class PlainTableBuilder: public TableBuilder {
std::vector<std::unique_ptr<IntTblPropCollector>>
table_properties_collectors_;
BloomBlockBuilder bloom_block_;
std::unique_ptr<PlainTableIndexBuilder> index_builder_;
WritableFileWriter* file_;
uint64_t offset_ = 0;
uint32_t bloom_bits_per_key_;
size_t huge_page_tlb_size_;
Status status_;
TableProperties properties_;
PlainTableKeyEncoder encoder_;
bool store_index_in_file_;
std::vector<uint32_t> keys_or_prefixes_hashes_;
bool closed_ = false; // Either Finish() or Abandon() has been called.
+16 -2
View File
@@ -42,8 +42,10 @@ TableBuilder* PlainTableFactory::NewTableBuilder(
table_builder_options.ioptions, table_builder_options.moptions,
table_builder_options.int_tbl_prop_collector_factories, column_family_id,
file, table_options_.user_key_len, table_options_.encoding_type,
table_options_.index_sparseness,
table_builder_options.column_family_name);
table_options_.index_sparseness, table_options_.bloom_bits_per_key,
table_builder_options.column_family_name, 6,
table_options_.huge_page_tlb_size, table_options_.hash_table_ratio,
table_options_.store_index_in_file);
}
std::string PlainTableFactory::GetPrintableTableOptions() const {
@@ -55,15 +57,27 @@ std::string PlainTableFactory::GetPrintableTableOptions() const {
snprintf(buffer, kBufferSize, " user_key_len: %u\n",
table_options_.user_key_len);
ret.append(buffer);
snprintf(buffer, kBufferSize, " bloom_bits_per_key: %d\n",
table_options_.bloom_bits_per_key);
ret.append(buffer);
snprintf(buffer, kBufferSize, " hash_table_ratio: %lf\n",
table_options_.hash_table_ratio);
ret.append(buffer);
snprintf(buffer, kBufferSize, " index_sparseness: %" ROCKSDB_PRIszt "\n",
table_options_.index_sparseness);
ret.append(buffer);
snprintf(buffer, kBufferSize, " huge_page_tlb_size: %" ROCKSDB_PRIszt "\n",
table_options_.huge_page_tlb_size);
ret.append(buffer);
snprintf(buffer, kBufferSize, " encoding_type: %d\n",
table_options_.encoding_type);
ret.append(buffer);
snprintf(buffer, kBufferSize, " full_scan_mode: %d\n",
table_options_.full_scan_mode);
ret.append(buffer);
snprintf(buffer, kBufferSize, " store_index_in_file: %d\n",
table_options_.store_index_in_file);
ret.append(buffer);
return ret;
}
+2 -1
View File
@@ -204,7 +204,8 @@ static std::unordered_map<std::string, OptionTypeInfo> plain_table_type_info = {
{offsetof(struct PlainTableOptions, full_scan_mode), OptionType::kBoolean,
OptionVerificationType::kNormal, false, 0}},
{"store_index_in_file",
{0, OptionType::kBoolean, OptionVerificationType::kDeprecated, false, 0}}};
{offsetof(struct PlainTableOptions, store_index_in_file),
OptionType::kBoolean, OptionVerificationType::kNormal, false, 0}}};
} // namespace rocksdb
#endif // ROCKSDB_LITE
+102 -22
View File
@@ -294,8 +294,47 @@ Status PlainTableReader::PopulateIndex(TableProperties* props,
assert(props != nullptr);
table_properties_.reset(props);
// index_in_file and bloom_in_file features are deprecated.
// Even if they exist in file, ignore them and always reconstruct.
BlockContents index_block_contents;
Status s = ReadMetaBlock(file_info_.file.get(), nullptr /* prefetch_buffer */,
file_size_, kPlainTableMagicNumber, ioptions_,
PlainTableIndexBuilder::kPlainTableIndexBlock,
&index_block_contents,
true /* compression_type_missing */);
bool index_in_file = s.ok();
BlockContents bloom_block_contents;
bool bloom_in_file = false;
// We only need to read the bloom block if index block is in file.
if (index_in_file) {
s = ReadMetaBlock(file_info_.file.get(), nullptr /* prefetch_buffer */,
file_size_, kPlainTableMagicNumber, ioptions_,
BloomBlockBuilder::kBloomBlock, &bloom_block_contents,
true /* compression_type_missing */);
bloom_in_file = s.ok() && bloom_block_contents.data.size() > 0;
}
Slice* bloom_block;
if (bloom_in_file) {
// If bloom_block_contents.allocation is not empty (which will be the case
// for non-mmap mode), it holds the alloated memory for the bloom block.
// It needs to be kept alive to keep `bloom_block` valid.
bloom_block_alloc_ = std::move(bloom_block_contents.allocation);
bloom_block = &bloom_block_contents.data;
} else {
bloom_block = nullptr;
}
Slice* index_block;
if (index_in_file) {
// If index_block_contents.allocation is not empty (which will be the case
// for non-mmap mode), it holds the alloated memory for the index block.
// It needs to be kept alive to keep `index_block` valid.
index_block_alloc_ = std::move(index_block_contents.allocation);
index_block = &index_block_contents.data;
} else {
index_block = nullptr;
}
if ((prefix_extractor_ == nullptr) && (hash_table_ratio != 0)) {
// moptions.prefix_extractor is requried for a hash-based look-up.
@@ -308,36 +347,77 @@ Status PlainTableReader::PopulateIndex(TableProperties* props,
// offset) and append it to IndexRecordList, which is a data structure created
// to store them.
// Allocate bloom filter here for total order mode.
if (IsTotalOrderMode()) {
uint32_t num_bloom_bits =
static_cast<uint32_t>(table_properties_->num_entries) *
bloom_bits_per_key;
if (num_bloom_bits > 0) {
enable_bloom_ = true;
bloom_.SetTotalBits(&arena_, num_bloom_bits, ioptions_.bloom_locality,
huge_page_tlb_size, ioptions_.info_log);
if (!index_in_file) {
// Allocate bloom filter here for total order mode.
if (IsTotalOrderMode()) {
uint32_t num_bloom_bits =
static_cast<uint32_t>(table_properties_->num_entries) *
bloom_bits_per_key;
if (num_bloom_bits > 0) {
enable_bloom_ = true;
bloom_.SetTotalBits(&arena_, num_bloom_bits, ioptions_.bloom_locality,
huge_page_tlb_size, ioptions_.info_log);
}
}
} else if (bloom_in_file) {
enable_bloom_ = true;
auto num_blocks_property = props->user_collected_properties.find(
PlainTablePropertyNames::kNumBloomBlocks);
uint32_t num_blocks = 0;
if (num_blocks_property != props->user_collected_properties.end()) {
Slice temp_slice(num_blocks_property->second);
if (!GetVarint32(&temp_slice, &num_blocks)) {
num_blocks = 0;
}
}
// cast away const qualifier, because bloom_ won't be changed
bloom_.SetRawData(
const_cast<unsigned char*>(
reinterpret_cast<const unsigned char*>(bloom_block->data())),
static_cast<uint32_t>(bloom_block->size()) * 8, num_blocks);
} else {
// Index in file but no bloom in file. Disable bloom filter in this case.
enable_bloom_ = false;
bloom_bits_per_key = 0;
}
PlainTableIndexBuilder index_builder(&arena_, ioptions_, prefix_extractor_,
index_sparseness, hash_table_ratio,
huge_page_tlb_size);
std::vector<uint32_t> prefix_hashes;
Status s = PopulateIndexRecordList(&index_builder, &prefix_hashes);
if (!s.ok()) {
return s;
if (!index_in_file) {
s = PopulateIndexRecordList(&index_builder, &prefix_hashes);
if (!s.ok()) {
return s;
}
} else {
s = index_.InitFromRawData(*index_block);
if (!s.ok()) {
return s;
}
}
if (!index_in_file) {
// Calculated bloom filter size and allocate memory for
// bloom filter based on the number of prefixes, then fill it.
AllocateAndFillBloom(bloom_bits_per_key, index_.GetNumPrefixes(),
huge_page_tlb_size, &prefix_hashes);
}
// Calculated bloom filter size and allocate memory for
// bloom filter based on the number of prefixes, then fill it.
AllocateAndFillBloom(bloom_bits_per_key, index_.GetNumPrefixes(),
huge_page_tlb_size, &prefix_hashes);
// Fill two table properties.
props->user_collected_properties["plain_table_hash_table_size"] =
ToString(index_.GetIndexSize() * PlainTableIndex::kOffsetLen);
props->user_collected_properties["plain_table_sub_index_size"] =
ToString(index_.GetSubIndexSize());
if (!index_in_file) {
props->user_collected_properties["plain_table_hash_table_size"] =
ToString(index_.GetIndexSize() * PlainTableIndex::kOffsetLen);
props->user_collected_properties["plain_table_sub_index_size"] =
ToString(index_.GetSubIndexSize());
} else {
props->user_collected_properties["plain_table_hash_table_size"] =
ToString(0);
props->user_collected_properties["plain_table_sub_index_size"] =
ToString(0);
}
return Status::OK();
}
+3 -3
View File
@@ -237,9 +237,9 @@ Status SstFileWriter::Open(const std::string& file_path) {
&int_tbl_prop_collector_factories, compression_type, compression_opts,
nullptr /* compression_dict */, r->skip_filters, r->column_family_name,
unknown_level);
r->file_writer.reset(new WritableFileWriter(
std::move(sst_file), file_path, r->env_options, r->ioptions.env,
nullptr /* stats */, r->ioptions.listeners));
r->file_writer.reset(
new WritableFileWriter(std::move(sst_file), file_path, r->env_options,
nullptr /* stats */, r->ioptions.listeners));
// TODO(tec) : If table_factory is using compressed block cache, we will
// be adding the external sst file blocks into it, which is wasteful.
+7 -9
View File
@@ -4667,25 +4667,23 @@ void VerifyDBFromDB(std::string& truth_db_name) {
int64_t seek = 0;
int64_t seek_found = 0;
int64_t bytes = 0;
const int64_t default_value_max = 64 * 1024 * 1024;
int64_t value_max = default_value_max;
int64_t value_max = FLAGS_mix_max_value_size;
int64_t scan_len_max = FLAGS_mix_max_scan_len;
double write_rate = 1000000.0;
double read_rate = 1000000.0;
std::vector<double> ratio{FLAGS_mix_get_ratio, FLAGS_mix_put_ratio,
FLAGS_mix_seek_ratio};
char value_buffer[default_value_max];
std::vector<double> ratio;
char value_buffer[2 * value_max];
QueryDecider query;
RandomGenerator gen;
Status s;
if (value_max > FLAGS_mix_max_value_size) {
value_max = FLAGS_mix_max_value_size;
}
ReadOptions options(FLAGS_verify_checksum, true);
std::unique_ptr<const char[]> key_guard;
Slice key = AllocateKey(&key_guard);
PinnableSlice pinnable_val;
ratio.push_back(FLAGS_mix_get_ratio);
ratio.push_back(FLAGS_mix_put_ratio);
ratio.push_back(FLAGS_mix_seek_ratio);
query.Initiate(ratio);
// the limit of qps initiation
@@ -4819,7 +4817,7 @@ void VerifyDBFromDB(std::string& truth_db_name) {
}
}
}
char msg[256];
char msg[100];
snprintf(msg, sizeof(msg),
"( Gets:%" PRIu64 " Puts:%" PRIu64 " Seek:%" PRIu64 " of %" PRIu64
" in %" PRIu64 " found)\n",
-2
View File
@@ -11,7 +11,6 @@ namespace rocksdb {
void CompactionJobStats::Reset() {
elapsed_micros = 0;
cpu_micros = 0;
num_input_records = 0;
num_input_files = 0;
@@ -46,7 +45,6 @@ void CompactionJobStats::Reset() {
void CompactionJobStats::Add(const CompactionJobStats& stats) {
elapsed_micros += stats.elapsed_micros;
cpu_micros += stats.cpu_micros;
num_input_records += stats.num_input_records;
num_input_files += stats.num_input_files;
+6 -2
View File
@@ -195,7 +195,9 @@ struct CompressionDict {
}
#if ZSTD_VERSION_NUMBER >= 700
const ZSTD_CDict* GetDigestedZstdCDict() const { return zstd_cdict_; }
const ZSTD_CDict* GetDigestedZstdCDict() const {
return zstd_cdict_;
}
#endif // ZSTD_VERSION_NUMBER >= 700
Slice GetRawDict() const { return dict_; }
@@ -257,7 +259,9 @@ struct UncompressionDict {
}
#ifdef ROCKSDB_ZSTD_DDICT
const ZSTD_DDict* GetDigestedZstdDDict() const { return zstd_ddict_; }
const ZSTD_DDict* GetDigestedZstdDDict() const {
return zstd_ddict_;
}
#endif // ROCKSDB_ZSTD_DDICT
Slice GetRawDict() const { return dict_; }
+3 -4
View File
@@ -52,12 +52,11 @@ DeleteScheduler::~DeleteScheduler() {
}
Status DeleteScheduler::DeleteFile(const std::string& file_path,
const std::string& dir_to_sync,
const bool force_bg) {
const std::string& dir_to_sync) {
Status s;
if (rate_bytes_per_sec_.load() <= 0 || (!force_bg &&
if (rate_bytes_per_sec_.load() <= 0 ||
total_trash_size_.load() >
sst_file_manager_->GetTotalSize() * max_trash_db_ratio_.load())) {
sst_file_manager_->GetTotalSize() * max_trash_db_ratio_.load()) {
// Rate limiting is disabled or trash size makes up more than
// max_trash_db_ratio_ (default 25%) of the total DB size
TEST_SYNC_POINT("DeleteScheduler::DeleteFile");
+2 -5
View File
@@ -46,11 +46,8 @@ class DeleteScheduler {
rate_bytes_per_sec_.store(bytes_per_sec);
}
// Mark file as trash directory and schedule it's deletion. If force_bg is
// set, it forces the file to always be deleted in the background thread,
// except when rate limiting is disabled
Status DeleteFile(const std::string& fname, const std::string& dir_to_sync,
const bool force_bg = false);
// Mark file as trash directory and schedule it's deletion
Status DeleteFile(const std::string& fname, const std::string& dir_to_sync);
// Wait for all files being deleteing in the background to finish or for
// destructor to be called.
+4 -21
View File
@@ -77,7 +77,6 @@ Status RandomAccessFileReader::Read(uint64_t offset, size_t n, Slice* result,
StopWatch sw(env_, stats_, hist_type_,
(stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
true /*delay_enabled*/);
auto prev_perf_level = GetPerfLevel();
IOSTATS_TIMER_GUARD(read_nanos);
if (use_direct_io()) {
#ifndef ROCKSDB_LITE
@@ -106,11 +105,8 @@ Status RandomAccessFileReader::Read(uint64_t offset, size_t n, Slice* result,
start_ts = std::chrono::system_clock::now();
orig_offset = aligned_offset + buf.CurrentSize();
}
{
IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, env_);
s = file_->Read(aligned_offset + buf.CurrentSize(), allowed, &tmp,
buf.Destination());
}
s = file_->Read(aligned_offset + buf.CurrentSize(), allowed, &tmp,
buf.Destination());
if (ShouldNotifyListeners()) {
auto finish_ts = std::chrono::system_clock::now();
NotifyOnFileReadFinish(orig_offset, tmp.size(), start_ts, finish_ts,
@@ -155,10 +151,7 @@ Status RandomAccessFileReader::Read(uint64_t offset, size_t n, Slice* result,
start_ts = std::chrono::system_clock::now();
}
#endif
{
IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, env_);
s = file_->Read(offset + pos, allowed, &tmp_result, scratch + pos);
}
s = file_->Read(offset + pos, allowed, &tmp_result, scratch + pos);
#ifndef ROCKSDB_LITE
if (ShouldNotifyListeners()) {
auto finish_ts = std::chrono::system_clock::now();
@@ -183,12 +176,10 @@ Status RandomAccessFileReader::Read(uint64_t offset, size_t n, Slice* result,
*result = Slice(res_scratch, s.ok() ? pos : 0);
}
IOSTATS_ADD_IF_POSITIVE(bytes_read, result->size());
SetPerfLevel(prev_perf_level);
}
if (stats_ != nullptr && file_read_hist_ != nullptr) {
file_read_hist_->Add(elapsed);
}
return s;
}
@@ -418,14 +409,11 @@ Status WritableFileWriter::SyncInternal(bool use_fsync) {
Status s;
IOSTATS_TIMER_GUARD(fsync_nanos);
TEST_SYNC_POINT("WritableFileWriter::SyncInternal:0");
auto prev_perf_level = GetPerfLevel();
IOSTATS_CPU_TIMER_GUARD(cpu_write_nanos, env_);
if (use_fsync) {
s = writable_file_->Fsync();
} else {
s = writable_file_->Sync();
}
SetPerfLevel(prev_perf_level);
return s;
}
@@ -465,12 +453,7 @@ Status WritableFileWriter::WriteBuffered(const char* data, size_t size) {
old_size = next_write_offset_;
}
#endif
{
auto prev_perf_level = GetPerfLevel();
IOSTATS_CPU_TIMER_GUARD(cpu_write_nanos, env_);
s = writable_file_->Append(Slice(src, allowed));
SetPerfLevel(prev_perf_level);
}
s = writable_file_->Append(Slice(src, allowed));
#ifndef ROCKSDB_LITE
if (ShouldNotifyListeners()) {
auto finish_ts = std::chrono::system_clock::now();
+1 -4
View File
@@ -174,7 +174,6 @@ class WritableFileWriter {
std::unique_ptr<WritableFile> writable_file_;
std::string file_name_;
Env* env_;
AlignedBuffer buf_;
size_t max_buffer_size_;
// Actually written data size can be used for truncate
@@ -196,12 +195,10 @@ class WritableFileWriter {
public:
WritableFileWriter(
std::unique_ptr<WritableFile>&& file, const std::string& _file_name,
const EnvOptions& options, Env* env = nullptr,
Statistics* stats = nullptr,
const EnvOptions& options, Statistics* stats = nullptr,
const std::vector<std::shared_ptr<EventListener>>& listeners = {})
: writable_file_(std::move(file)),
file_name_(_file_name),
env_(env),
buf_(),
max_buffer_size_(options.writable_file_max_buffer_size),
filesize_(0),
+1 -8
View File
@@ -89,23 +89,16 @@ Status CreateFile(Env* env, const std::string& destination,
Status DeleteSSTFile(const ImmutableDBOptions* db_options,
const std::string& fname, const std::string& dir_to_sync) {
return DeleteDBFile(db_options, fname, dir_to_sync, false);
}
Status DeleteDBFile(const ImmutableDBOptions* db_options,
const std::string& fname, const std::string& dir_to_sync,
const bool force_bg) {
#ifndef ROCKSDB_LITE
auto sfm =
static_cast<SstFileManagerImpl*>(db_options->sst_file_manager.get());
if (sfm) {
return sfm->ScheduleFileDeletion(fname, dir_to_sync, force_bg);
return sfm->ScheduleFileDeletion(fname, dir_to_sync);
} else {
return db_options->env->DeleteFile(fname);
}
#else
(void)dir_to_sync;
(void)force_bg;
// SstFileManager is not supported in ROCKSDB_LITE
return db_options->env->DeleteFile(fname);
#endif
-5
View File
@@ -25,9 +25,4 @@ extern Status DeleteSSTFile(const ImmutableDBOptions* db_options,
const std::string& fname,
const std::string& path_to_sync);
extern Status DeleteDBFile(const ImmutableDBOptions* db_options,
const std::string& fname,
const std::string& path_to_sync,
const bool force_bg);
} // namespace rocksdb
+2 -4
View File
@@ -402,11 +402,9 @@ bool SstFileManagerImpl::CancelErrorRecovery(ErrorHandler* handler) {
}
Status SstFileManagerImpl::ScheduleFileDeletion(
const std::string& file_path, const std::string& path_to_sync,
const bool force_bg) {
const std::string& file_path, const std::string& path_to_sync) {
TEST_SYNC_POINT("SstFileManagerImpl::ScheduleFileDeletion");
return delete_scheduler_.DeleteFile(file_path, path_to_sync,
force_bg);
return delete_scheduler_.DeleteFile(file_path, path_to_sync);
}
void SstFileManagerImpl::WaitForEmptyTrash() {
+2 -5
View File
@@ -111,12 +111,9 @@ class SstFileManagerImpl : public SstFileManager {
// not guaranteed
bool CancelErrorRecovery(ErrorHandler* db);
// Mark file as trash and schedule it's deletion. If force_bg is set, it
// forces the file to be deleting in the background regardless of DB size,
// except when rate limited delete is disabled
// Mark file as trash and schedule it's deletion.
virtual Status ScheduleFileDeletion(const std::string& file_path,
const std::string& dir_to_sync,
const bool force_bg = false);
const std::string& dir_to_sync);
// Wait for all files being deleteing in the background to finish or for
// destructor to be called.
+3 -3
View File
@@ -1743,8 +1743,8 @@ std::pair<bool, int64_t> BlobDBImpl::DeleteObsoleteFiles(bool aborted) {
bfile->PathName().c_str());
blob_files_.erase(bfile->BlobFileNumber());
Status s = DeleteDBFile(&(db_impl_->immutable_db_options()),
bfile->PathName(), blob_dir_, true);
Status s = DeleteSSTFile(&(db_impl_->immutable_db_options()),
bfile->PathName(), blob_dir_);
if (!s.ok()) {
ROCKS_LOG_ERROR(db_options_.info_log,
"File failed to be deleted as obsolete %s",
@@ -1834,7 +1834,7 @@ Status DestroyBlobDB(const std::string& dbname, const Options& options,
uint64_t number;
FileType type;
if (ParseFileName(f, &number, &type) && type == kBlobFile) {
Status del = DeleteDBFile(&soptions, blobdir + "/" + f, blobdir, true);
Status del = DeleteSSTFile(&soptions, blobdir + "/" + f, blobdir);
if (status.ok() && !del.ok()) {
status = del;
}
@@ -791,15 +791,14 @@ class WritePreparedCommitEntryPreReleaseCallback : public PreReleaseCallback {
bool publish_seq_;
};
// For two_write_queues commit both the aborted batch and the cleanup batch and
// then published the seq
// For two_write_queues commit both the aborted batch and the cleanup batch and then published the seq
class WritePreparedRollbackPreReleaseCallback : public PreReleaseCallback {
public:
WritePreparedRollbackPreReleaseCallback(WritePreparedTxnDB* db,
DBImpl* db_impl,
SequenceNumber prep_seq,
SequenceNumber rollback_seq,
size_t prep_batch_cnt)
DBImpl* db_impl,
SequenceNumber prep_seq,
SequenceNumber rollback_seq,
size_t prep_batch_cnt)
: db_(db),
db_impl_(db_impl),
prep_seq_(prep_seq),