make L0 index/filter pinned memory usage predictable (#6911 )

Summary: Memory pinned by `pin_l0_filter_and_index_blocks_in_cache` needs to be predictable based on user config. This PR makes sure we do not pin extra memory for large files generated by intra-L0 (see https://github.com/facebook/rocksdb/issues/6889). Pull Request resolved: https://github.com/facebook/rocksdb/pull/6911 Test Plan: unit test Reviewed By: siying Differential Revision: D21835818 Pulled By: ajkr fbshipit-source-id: a11a088549d06bed8aacc2548d266e5983f0ead4
Move blob_log_{format,reader,writer}.{cc,h} to db/blob/ (#6960 )
2026-11-07 03:22:34 -05:00 · 2020-06-09 16:51:23 -07:00 · 2020-06-09 15:16:05 -07:00 · 2020-06-09 10:04:01 -07:00 · 2020-06-08 19:34:31 -07:00 · 2020-06-08 16:11:21 -07:00
76 changed files with 988 additions and 335 deletions
@@ -11,7 +11,59 @@ executors:
      shell: bash.exe

 jobs:
-  build:
+  build-linux:
+    machine:
+      image: ubuntu-1604:201903-01
+    resource_class: 2xlarge
+    steps:
+      - checkout # check out the code in the project directory
+      - run: pyenv global 3.5.2
+      - run: sudo apt-get update -y
+      - run: sudo apt-get install -y libgflags-dev
+      - run: SKIP_FORMAT_BUCK_CHECKS=1 PRINT_PARALLEL_OUTPUTS=1 make J=32 all check -j32
+
+  build-linux-release:
+    machine:
+      image: ubuntu-1604:201903-01
+    resource_class: 2xlarge
+    steps:
+      - checkout # check out the code in the project directory
+      - run: make release -j32
+
+  build-linux-lite:
+    machine:
+      image: ubuntu-1604:201903-01 
+    resource_class: 2xlarge
+    steps:
+      - checkout # check out the code in the project directory
+      - run: pyenv global 3.5.2
+      - run: SKIP_FORMAT_BUCK_CHECKS=1 PRINT_PARALLEL_OUTPUTS=1 LITE=1 make J=32 all check -j32
+
+  build-linux-lite-release:
+    machine:
+      image: ubuntu-1604:201903-01
+    resource_class: large
+    steps:
+      - checkout # check out the code in the project directory
+      - run: make release -j32
+
+  build-linux-clang-no-test:
+    machine:
+      image: ubuntu-1604:201903-01
+    resource_class: 2xlarge
+    steps:
+      - checkout # check out the code in the project directory
+      - run: USE_CLANG=1 make all -j32
+
+  build-linux-cmake:
+    machine:
+      image: ubuntu-1604:201903-01
+    resource_class: 2xlarge
+    steps:
+      - checkout # check out the code in the project directory
+      - run: mkdir build && cd build && cmake -DWITH_GFLAGS=0 .. && make -j32
+
+  build-windows:
    executor: windows-2xlarge
    
    environment:
@@ -54,3 +106,26 @@ jobs:
          shell: powershell.exe
          command: |
            build_tools\run_ci_db_test.ps1 -SuiteRun db_basic_test,db_test,db_test2,env_basic_test,env_test,db_merge_operand_test -Concurrency 16
+
+workflows:
+  build-linux:
+    jobs:
+      - build-linux
+  build-linux-lite:
+    jobs:
+      - build-linux-lite
+  build-linux-release:
+    jobs:
+      - build-linux-release
+  build-linux-lite-release:
+    jobs:
+      - build-linux-lite-release
+  build-linux-clang-no-test:
+    jobs:
+      - build-linux-clang-no-test
+  build-linux-cmake:
+    jobs:
+      - build-linux-cmake
+  build-windows:
+    jobs:
+      - build-windows
@@ -539,6 +539,9 @@ set(SOURCES
        db/blob/blob_file_addition.cc
        db/blob/blob_file_garbage.cc
        db/blob/blob_file_meta.cc
+        db/blob/blob_log_format.cc
+        db/blob/blob_log_reader.cc
+        db/blob/blob_log_writer.cc
        db/builder.cc
        db/c.cc
        db/column_family.cc
@@ -730,9 +733,6 @@ set(SOURCES
        utilities/blob_db/blob_db_impl_filesnapshot.cc
        utilities/blob_db/blob_dump_tool.cc
        utilities/blob_db/blob_file.cc
-        utilities/blob_db/blob_log_reader.cc
-        utilities/blob_db/blob_log_writer.cc
-        utilities/blob_db/blob_log_format.cc
        utilities/cassandra/cassandra_compaction_filter.cc
        utilities/cassandra/format.cc
        utilities/cassandra/merge_operator.cc
@@ -1090,6 +1090,7 @@ if(WITH_TESTS)
        table/sst_file_reader_test.cc
        table/table_test.cc
        table/block_fetcher_test.cc
+        test_util/testutil_test.cc
        tools/block_cache_analyzer/block_cache_trace_analyzer_test.cc
        tools/ldb_cmd_test.cc
        tools/reduce_levels_test.cc
@@ -2,6 +2,7 @@
 ## Unreleased
 ### Behavior Changes
 * Disable delete triggered compaction (NewCompactOnDeletionCollectorFactory) in universal compaction mode and num_levels = 1 in order to avoid a corruption bug.
+* `pin_l0_filter_and_index_blocks_in_cache` no longer applies to L0 files larger than `1.5 * write_buffer_size` to give more predictable memory usage. Such L0 files may exist due to intra-L0 compaction, external file ingestion, or user dynamically changing `write_buffer_size` (note, however, that files that are already pinned will continue being pinned, even after such a dynamic change).

 ### Bug Fixes
 * Fix consistency checking error swallowing in some cases when options.force_consistency_checks = true.
@@ -12,6 +13,9 @@
 * Fix corrupt key read from ingested file when iterator direction switches from reverse to forward at a key that is a prefix of another key in the same file. It is only possible in files with a non-zero global seqno.
 * Fix abnormally large estimate from GetApproximateSizes when a range starts near the end of one SST file and near the beginning of another. Now GetApproximateSizes consistently and fairly includes the size of SST metadata in addition to data blocks, attributing metadata proportionally among the data blocks based on their size.
 * Fix potential file descriptor leakage in PosixEnv's IsDirectory() and NewRandomAccessFile().
+* Fix false negative from the VerifyChecksum() API when there is a checksum mismatch in an index partition block in a BlockBasedTable format table file (index_type is kTwoLevelIndexSearch).
+* Fix sst_dump to return non-zero exit code if the specified file is not a recognized SST file or fails requested checks.
+* Fix incorrect results from batched MultiGet for duplicate keys, when the duplicate key matches the largest key of an SST file and the value type for the key in the file is a merge value.

 ### Public API Change
 * Flush(..., column_family) may return Status::ColumnFamilyDropped() instead of Status::InvalidArgument() if column_family is dropped while processing the flush request.
@@ -639,6 +639,7 @@ TESTS = \
 	blob_file_garbage_test \
 	timer_test \
 	db_with_timestamp_compaction_test \
+	testutil_test \

 PARALLEL_TEST = \
 	backupable_db_test \
@@ -975,6 +976,12 @@ J ?= 100%
 # Use this regexp to select the subset of tests whose names match.
 tests-regexp = .

+ifeq ($(PRINT_PARALLEL_OUTPUTS), 1)	
+	parallel_com = '{}'
+else
+	parallel_com = '{} >& t/log-{/}'
+endif
+
 .PHONY: check_0
 check_0:
 	$(AM_V_GEN)export TEST_TMPDIR=$(TMPD); \
@@ -988,7 +995,10 @@ check_0:
 	} \
 	  | $(prioritize_long_running_tests)				\
 	  | grep -E '$(tests-regexp)'					\
-	  | build_tools/gnu_parallel -j$(J) --plain --joblog=LOG $$eta --gnu '{} >& t/log-{/}'
+	  | build_tools/gnu_parallel -j$(J) --plain --joblog=LOG $$eta --gnu  $(parallel_com) ; \
+	parallel_retcode=$$? ; \
+	awk '{ if ($$7 != 0 || $$8 != 0) { if ($$7 == "Exitval") { h = $$0; } else { if (!f) print h; print; f = 1 } } } END { if(f) exit 1; }' < LOG ; \
+	if [ $$parallel_retcode -ne 0 ] ; then exit 1 ; fi

 valgrind-blacklist-regexp = InlineSkipTest.ConcurrentInsert|TransactionStressTest.DeadlockStress|DBCompactionTest.SuggestCompactRangeNoTwoLevel0Compactions|BackupableDBTest.RateLimiting|DBTest.CloseSpeedup|DBTest.ThreadStatusFlush|DBTest.RateLimitingTest|DBTest.EncodeDecompressedBlockSizeTest|FaultInjectionTest.UninstalledCompaction|HarnessTest.Randomized|ExternalSSTFileTest.CompactDuringAddFileRandom|ExternalSSTFileTest.IngestFileWithGlobalSeqnoRandomized|MySQLStyleTransactionTest.TransactionStressTest

@@ -1047,9 +1057,11 @@ ifndef ASSERT_STATUS_CHECKED # not yet working with these tests
 	sh tools/rocksdb_dump_test.sh
 endif
 endif
-endif
+endif	
+ifndef SKIP_FORMAT_BUCK_CHECKS
 	$(MAKE) check-format
 	$(MAKE) check-buck-targets
+endif

 # TODO add ldb_tests
 check_some: $(SUBSET)
@@ -1866,6 +1878,9 @@ blob_file_garbage_test: db/blob/blob_file_garbage_test.o $(LIBOBJECTS) $(TESTHAR
 timer_test: util/timer_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)

+testutil_test: test_util/testutil_test.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(AM_LINK)
+
 #-------------------------------------------------
 # make install related stuff
 INSTALL_PATH ?= /usr/local
@@ -78,7 +78,6 @@ ROCKSDB_PREPROCESSOR_FLAGS = [
    # Directories with files for #include
    "-I" + REPO_PATH + "include/",
    "-I" + REPO_PATH,
-    "-I" + REPO_PATH + "third-party/gtest-1.8.1/fused-src/",
 ]

 ROCKSDB_ARCH_PREPROCESSOR_FLAGS = {
@@ -125,6 +124,9 @@ cpp_library(
        "db/blob/blob_file_addition.cc",
        "db/blob/blob_file_garbage.cc",
        "db/blob/blob_file_meta.cc",
+        "db/blob/blob_log_format.cc",
+        "db/blob/blob_log_reader.cc",
+        "db/blob/blob_log_writer.cc",
        "db/builder.cc",
        "db/c.cc",
        "db/column_family.cc",
@@ -319,9 +321,6 @@ cpp_library(
        "utilities/blob_db/blob_db_impl_filesnapshot.cc",
        "utilities/blob_db/blob_dump_tool.cc",
        "utilities/blob_db/blob_file.cc",
-        "utilities/blob_db/blob_log_format.cc",
-        "utilities/blob_db/blob_log_reader.cc",
-        "utilities/blob_db/blob_log_writer.cc",
        "utilities/cassandra/cassandra_compaction_filter.cc",
        "utilities/cassandra/format.cc",
        "utilities/cassandra/merge_operator.cc",
@@ -398,11 +397,10 @@ cpp_library(
    os_deps = ROCKSDB_OS_DEPS,
    os_preprocessor_flags = ROCKSDB_OS_PREPROCESSOR_FLAGS,
    preprocessor_flags = ROCKSDB_PREPROCESSOR_FLAGS,
-    deps = [
-        ":rocksdb_lib",
-        ":rocksdb_third_party_gtest",
+    deps = [":rocksdb_lib"],
+    external_deps = ROCKSDB_EXTERNAL_DEPS + [
+        ("googletest", None, "gtest"),
    ],
-    external_deps = ROCKSDB_EXTERNAL_DEPS,
 )

 cpp_library(
@@ -449,19 +447,6 @@ cpp_library(
    external_deps = ROCKSDB_EXTERNAL_DEPS,
 )

-cpp_library(
-    name = "rocksdb_third_party_gtest",
-    srcs = ["third-party/gtest-1.8.1/fused-src/gtest/gtest-all.cc"],
-    headers = ["third-party/gtest-1.8.1/fused-src/gtest/gtest.h"],
-    arch_preprocessor_flags = ROCKSDB_ARCH_PREPROCESSOR_FLAGS,
-    compiler_flags = ROCKSDB_COMPILER_FLAGS,
-    os_deps = ROCKSDB_OS_DEPS,
-    os_preprocessor_flags = ROCKSDB_OS_PREPROCESSOR_FLAGS,
-    preprocessor_flags = ROCKSDB_PREPROCESSOR_FLAGS,
-    deps = [],
-    external_deps = ROCKSDB_EXTERNAL_DEPS,
-)
-
 cpp_library(
    name = "env_basic_test_lib",
    srcs = ["env/env_basic_test.cc"],
@@ -1485,6 +1470,13 @@ ROCKS_TESTS = [
        [],
        [],
    ],
+    [
+        "testutil_test",
+        "test_util/testutil_test.cc",
+        "serial",
+        [],
+        [],
+    ],
    [
        "thread_list_test",
        "util/thread_list_test.cc",
@@ -1638,18 +1630,17 @@ ROCKS_TESTS = [
 # Do not build the tests in opt mode, since SyncPoint and other test code
 # will not be included.
 [
-    test_binary(
-        extra_compiler_flags = extra_compiler_flags,
-        extra_deps = extra_deps,
-        parallelism = parallelism,
-        rocksdb_arch_preprocessor_flags = ROCKSDB_ARCH_PREPROCESSOR_FLAGS,
-        rocksdb_compiler_flags = ROCKSDB_COMPILER_FLAGS,
-        rocksdb_external_deps = ROCKSDB_EXTERNAL_DEPS,
-        rocksdb_os_deps = ROCKSDB_OS_DEPS,
-        rocksdb_os_preprocessor_flags = ROCKSDB_OS_PREPROCESSOR_FLAGS,
-        rocksdb_preprocessor_flags = ROCKSDB_PREPROCESSOR_FLAGS,
-        test_cc = test_cc,
-        test_name = test_name,
+    cpp_unittest(
+        name = test_name,
+        srcs = [test_cc],
+        arch_preprocessor_flags = ROCKSDB_ARCH_PREPROCESSOR_FLAGS,
+        os_preprocessor_flags = ROCKSDB_OS_PREPROCESSOR_FLAGS,
+        compiler_flags = ROCKSDB_COMPILER_FLAGS,
+        preprocessor_flags = ROCKSDB_PREPROCESSOR_FLAGS,
+        deps = [":rocksdb_test_lib"] + extra_deps,
+        external_deps = ROCKSDB_EXTERNAL_DEPS + [
+            ("googletest", None, "gtest"),
+        ],
    )
    for test_name, test_cc, parallelism, extra_deps, extra_compiler_flags in ROCKS_TESTS
    if not is_opt_mode
@@ -108,16 +108,6 @@ def get_tests(repo_path):

    return tests

-
-# Get gtest dir from Makefile
-def get_gtest_dir(repo_path):
-    for line in open(repo_path + "/Makefile"):
-        if line.strip().startswith("GTEST_DIR ="):
-            return line.split("=")[1].strip()
-    # if not found
-    exit_with_error("Unable to find GTEST_DIR in Makefile")
-
-
 # Parse extra dependencies passed by user from command line
 def get_dependencies():
    deps_map = {
@@ -151,13 +141,11 @@ def generate_targets(repo_path, deps_map):
    cc_files = get_cc_files(repo_path)
    # get tests from Makefile
    tests = get_tests(repo_path)
-    # get gtest dir
-    gtest_dir = get_gtest_dir(repo_path) + "/"

    if src_mk is None or cc_files is None or tests is None:
        return False

-    TARGETS = TARGETSBuilder("%s/TARGETS" % repo_path, gtest_dir)
+    TARGETS = TARGETSBuilder("%s/TARGETS" % repo_path)

    # rocksdb_lib
    TARGETS.add_library(
@@ -171,7 +159,10 @@ def generate_targets(repo_path, deps_map):
        src_mk.get("TEST_LIB_SOURCES", []) +
        src_mk.get("EXP_LIB_SOURCES", []) +
        src_mk.get("ANALYZER_LIB_SOURCES", []),
-        [":rocksdb_lib", ":rocksdb_third_party_gtest"])
+        [":rocksdb_lib"],
+        extra_external_deps=""" + [
+        ("googletest", None, "gtest"),
+    ]""")
    # rocksdb_tools_lib
    TARGETS.add_library(
        "rocksdb_tools_lib",
@@ -185,12 +176,6 @@ def generate_targets(repo_path, deps_map):
        src_mk.get("ANALYZER_LIB_SOURCES", [])
        + src_mk.get('STRESS_LIB_SOURCES', [])
        + ["test_util/testutil.cc"])
-    # rocksdb_third_party_gtest
-    TARGETS.add_library(
-        "rocksdb_third_party_gtest",
-        [gtest_dir + "gtest/gtest-all.cc"],
-        [],
-        [gtest_dir + "gtest/gtest.h"])

    print("Extra dependencies:\n{0}".format(json.dumps(deps_map)))
    # test for every test we found in the Makefile
@@ -25,11 +25,10 @@ def pretty_list(lst, indent=8):


 class TARGETSBuilder(object):
-    def __init__(self, path, gtest_dir):
+    def __init__(self, path):
        self.path = path
        self.targets_file = open(path, 'w')
-        header = targets_cfg.rocksdb_target_header_template.format(
-            gtest_dir=gtest_dir)
+        header = targets_cfg.rocksdb_target_header_template
        self.targets_file.write(header)
        self.total_lib = 0
        self.total_bin = 0
@@ -39,7 +38,8 @@ class TARGETSBuilder(object):
    def __del__(self):
        self.targets_file.close()

-    def add_library(self, name, srcs, deps=None, headers=None):
+    def add_library(self, name, srcs, deps=None, headers=None,
+                    extra_external_deps=""):
        headers_attr_prefix = ""
        if headers is None:
            headers_attr_prefix = "auto_"
@@ -51,7 +51,8 @@ class TARGETSBuilder(object):
            srcs=pretty_list(srcs),
            headers_attr_prefix=headers_attr_prefix,
            headers=headers,
-            deps=pretty_list(deps)))
+            deps=pretty_list(deps),
+            extra_external_deps=extra_external_deps))
        self.total_lib = self.total_lib + 1

    def add_rocksdb_library(self, name, srcs, headers=None):
@@ -85,14 +85,13 @@ ROCKSDB_PREPROCESSOR_FLAGS = [
    # Directories with files for #include
    "-I" + REPO_PATH + "include/",
    "-I" + REPO_PATH,
-    "-I" + REPO_PATH + "{gtest_dir}",
 ]

-ROCKSDB_ARCH_PREPROCESSOR_FLAGS = {{
+ROCKSDB_ARCH_PREPROCESSOR_FLAGS = {
    "x86_64": [
        "-DHAVE_PCLMUL",
    ],
-}}
+}

 build_mode = read_config("fbcode", "build_mode")

@@ -134,7 +133,7 @@ cpp_library(
    os_preprocessor_flags = ROCKSDB_OS_PREPROCESSOR_FLAGS,
    preprocessor_flags = ROCKSDB_PREPROCESSOR_FLAGS,
    deps = [{deps}],
-    external_deps = ROCKSDB_EXTERNAL_DEPS,
+    external_deps = ROCKSDB_EXTERNAL_DEPS{extra_external_deps},
 )
 """

@@ -183,18 +182,17 @@ ROCKS_TESTS = [
 # Do not build the tests in opt mode, since SyncPoint and other test code
 # will not be included.
 [
-    test_binary(
-        extra_compiler_flags = extra_compiler_flags,
-        extra_deps = extra_deps,
-        parallelism = parallelism,
-        rocksdb_arch_preprocessor_flags = ROCKSDB_ARCH_PREPROCESSOR_FLAGS,
-        rocksdb_compiler_flags = ROCKSDB_COMPILER_FLAGS,
-        rocksdb_external_deps = ROCKSDB_EXTERNAL_DEPS,
-        rocksdb_os_deps = ROCKSDB_OS_DEPS,
-        rocksdb_os_preprocessor_flags = ROCKSDB_OS_PREPROCESSOR_FLAGS,
-        rocksdb_preprocessor_flags = ROCKSDB_PREPROCESSOR_FLAGS,
-        test_cc = test_cc,
-        test_name = test_name,
+    cpp_unittest(
+        name = test_name,
+        srcs = [test_cc],
+        arch_preprocessor_flags = ROCKSDB_ARCH_PREPROCESSOR_FLAGS,
+        os_preprocessor_flags = ROCKSDB_OS_PREPROCESSOR_FLAGS,
+        compiler_flags = ROCKSDB_COMPILER_FLAGS,
+        preprocessor_flags = ROCKSDB_PREPROCESSOR_FLAGS,
+        deps = [":rocksdb_test_lib"] + extra_deps,
+        external_deps = ROCKSDB_EXTERNAL_DEPS + [
+            ("googletest", None, "gtest"),
+        ],
    )
    for test_name, test_cc, parallelism, extra_deps, extra_compiler_flags in ROCKS_TESTS
    if not is_opt_mode
@@ -5,7 +5,7 @@
 //
 #ifndef ROCKSDB_LITE

-#include "utilities/blob_db/blob_log_format.h"
+#include "db/blob/blob_log_format.h"

 #include "util/coding.h"
 #include "util/crc32c.h"
@@ -5,7 +5,7 @@
 //
 #ifndef ROCKSDB_LITE

-#include "utilities/blob_db/blob_log_reader.h"
+#include "db/blob/blob_log_reader.h"

 #include <algorithm>

@@ -10,12 +10,12 @@
 #include <memory>
 #include <string>

+#include "db/blob/blob_log_format.h"
 #include "file/random_access_file_reader.h"
 #include "rocksdb/env.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/statistics.h"
 #include "rocksdb/status.h"
-#include "utilities/blob_db/blob_log_format.h"

 namespace ROCKSDB_NAMESPACE {

@@ -4,17 +4,17 @@
 //  (found in the LICENSE.Apache file in the root directory).
 #ifndef ROCKSDB_LITE

-#include "utilities/blob_db/blob_log_writer.h"
+#include "db/blob/blob_log_writer.h"

 #include <cstdint>
 #include <string>

+#include "db/blob/blob_log_format.h"
 #include "file/writable_file_writer.h"
 #include "monitoring/statistics.h"
 #include "rocksdb/env.h"
 #include "util/coding.h"
 #include "util/stop_watch.h"
-#include "utilities/blob_db/blob_log_format.h"

 namespace ROCKSDB_NAMESPACE {
 namespace blob_db {
@@ -10,12 +10,12 @@
 #include <memory>
 #include <string>

+#include "db/blob/blob_log_format.h"
 #include "rocksdb/env.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/statistics.h"
 #include "rocksdb/status.h"
 #include "rocksdb/types.h"
-#include "utilities/blob_db/blob_log_format.h"

 namespace ROCKSDB_NAMESPACE {

@@ -239,7 +239,9 @@ Status BuildTable(
          (internal_stats == nullptr) ? nullptr
                                      : internal_stats->GetFileReadHist(0),
          TableReaderCaller::kFlush, /*arena=*/nullptr,
-          /*skip_filter=*/false, level, /*smallest_compaction_key=*/nullptr,
+          /*skip_filter=*/false, level,
+          MaxFileSizeForL0MetaPin(mutable_cf_options),
+          /*smallest_compaction_key=*/nullptr,
          /*largest_compaction_key*/ nullptr,
          /*allow_unprepared_value*/ false));
      s = it->status();
@@ -654,6 +654,8 @@ Status CompactionJob::Run() {
                compact_->compaction->output_level()),
            TableReaderCaller::kCompactionRefill, /*arena=*/nullptr,
            /*skip_filters=*/false, compact_->compaction->output_level(),
+            MaxFileSizeForL0MetaPin(
+                *compact_->compaction->mutable_cf_options()),
            /*smallest_compaction_key=*/nullptr,
            /*largest_compaction_key=*/nullptr,
            /*allow_unprepared_value=*/false);
@@ -157,42 +157,6 @@ class CorruptionTest : public testing::Test {
    ASSERT_GE(max_expected, correct);
  }

-  void CorruptFile(const std::string& fname, int offset, int bytes_to_corrupt) {
-    struct stat sbuf;
-    if (stat(fname.c_str(), &sbuf) != 0) {
-      const char* msg = strerror(errno);
-      FAIL() << fname << ": " << msg;
-    }
-
-    if (offset < 0) {
-      // Relative to end of file; make it absolute
-      if (-offset > sbuf.st_size) {
-        offset = 0;
-      } else {
-        offset = static_cast<int>(sbuf.st_size + offset);
-      }
-    }
-    if (offset > sbuf.st_size) {
-      offset = static_cast<int>(sbuf.st_size);
-    }
-    if (offset + bytes_to_corrupt > sbuf.st_size) {
-      bytes_to_corrupt = static_cast<int>(sbuf.st_size - offset);
-    }
-
-    // Do it
-    std::string contents;
-    Status s = ReadFileToString(Env::Default(), fname, &contents);
-    ASSERT_TRUE(s.ok()) << s.ToString();
-    for (int i = 0; i < bytes_to_corrupt; i++) {
-      contents[i + offset] ^= 0x80;
-    }
-    s = WriteStringToFile(Env::Default(), contents, fname);
-    ASSERT_TRUE(s.ok()) << s.ToString();
-    Options options;
-    EnvOptions env_options;
-    ASSERT_NOK(VerifySstFileChecksum(options, env_options, fname));
-  }
-
  void Corrupt(FileType filetype, int offset, int bytes_to_corrupt) {
    // Pick file to corrupt
    std::vector<std::string> filenames;
@@ -211,7 +175,7 @@ class CorruptionTest : public testing::Test {
    }
    ASSERT_TRUE(!fname.empty()) << filetype;

-    CorruptFile(fname, offset, bytes_to_corrupt);
+    test::CorruptFile(fname, offset, bytes_to_corrupt);
  }

  // corrupts exactly one file at level `level`. if no file found at level,
@@ -221,7 +185,7 @@ class CorruptionTest : public testing::Test {
    db_->GetLiveFilesMetaData(&metadata);
    for (const auto& m : metadata) {
      if (m.level == level) {
-        CorruptFile(dbname_ + "/" + m.name, offset, bytes_to_corrupt);
+        test::CorruptFile(dbname_ + "/" + m.name, offset, bytes_to_corrupt);
        return;
      }
    }
@@ -556,7 +520,7 @@ TEST_F(CorruptionTest, RangeDeletionCorrupted) {
      ImmutableCFOptions(options_), kRangeDelBlock, &range_del_handle));

  ASSERT_OK(TryReopen());
-  CorruptFile(filename, static_cast<int>(range_del_handle.offset()), 1);
+  test::CorruptFile(filename, static_cast<int>(range_del_handle.offset()), 1);
  ASSERT_TRUE(TryReopen().IsCorruption());
 }

@@ -9,6 +9,7 @@

 #include "db/db_test_util.h"
 #include "port/stack_trace.h"
+#include "rocksdb/merge_operator.h"
 #include "rocksdb/perf_context.h"
 #include "rocksdb/utilities/debug.h"
 #include "table/block_based/block_based_table_reader.h"
@@ -17,6 +18,8 @@
 #if !defined(ROCKSDB_LITE)
 #include "test_util/sync_point.h"
 #endif
+#include "utilities/merge_operators.h"
+#include "utilities/merge_operators/string_append/stringappend.h"

 namespace ROCKSDB_NAMESPACE {

@@ -1340,6 +1343,57 @@ TEST_F(DBBasicTest, MultiGetBatchedSortedMultiFile) {
  } while (ChangeOptions());
 }

+TEST_F(DBBasicTest, MultiGetBatchedDuplicateKeys) {
+  Options opts = CurrentOptions();
+  opts.merge_operator = MergeOperators::CreateStringAppendOperator();
+  CreateAndReopenWithCF({"pikachu"}, opts);
+  SetPerfLevel(kEnableCount);
+  // To expand the power of this test, generate > 1 table file and
+  // mix with memtable
+  ASSERT_OK(Merge(1, "k1", "v1"));
+  ASSERT_OK(Merge(1, "k2", "v2"));
+  Flush(1);
+  MoveFilesToLevel(2, 1);
+  ASSERT_OK(Merge(1, "k3", "v3"));
+  ASSERT_OK(Merge(1, "k4", "v4"));
+  Flush(1);
+  MoveFilesToLevel(2, 1);
+  ASSERT_OK(Merge(1, "k4", "v4_2"));
+  ASSERT_OK(Merge(1, "k6", "v6"));
+  Flush(1);
+  MoveFilesToLevel(2, 1);
+  ASSERT_OK(Merge(1, "k7", "v7"));
+  ASSERT_OK(Merge(1, "k8", "v8"));
+  Flush(1);
+  MoveFilesToLevel(2, 1);
+
+  get_perf_context()->Reset();
+
+  std::vector<Slice> keys({"k8", "k8", "k8", "k4", "k4", "k1", "k3"});
+  std::vector<PinnableSlice> values(keys.size());
+  std::vector<ColumnFamilyHandle*> cfs(keys.size(), handles_[1]);
+  std::vector<Status> s(keys.size());
+
+  db_->MultiGet(ReadOptions(), handles_[1], keys.size(), keys.data(),
+                values.data(), s.data(), false);
+
+  ASSERT_EQ(values.size(), keys.size());
+  ASSERT_EQ(std::string(values[0].data(), values[0].size()), "v8");
+  ASSERT_EQ(std::string(values[1].data(), values[1].size()), "v8");
+  ASSERT_EQ(std::string(values[2].data(), values[2].size()), "v8");
+  ASSERT_EQ(std::string(values[3].data(), values[3].size()), "v4,v4_2");
+  ASSERT_EQ(std::string(values[4].data(), values[4].size()), "v4,v4_2");
+  ASSERT_EQ(std::string(values[5].data(), values[5].size()), "v1");
+  ASSERT_EQ(std::string(values[6].data(), values[6].size()), "v3");
+  ASSERT_EQ(24, (int)get_perf_context()->multiget_read_bytes);
+
+  for (Status& status : s) {
+    ASSERT_OK(status);
+  }
+
+  SetPerfLevel(kDisable);
+}
+
 TEST_F(DBBasicTest, MultiGetBatchedMultiLevel) {
  Options options = CurrentOptions();
  options.disable_auto_compactions = true;
@@ -3184,6 +3184,15 @@ TEST_P(DBCompactionTestWithParam, IntraL0Compaction) {
  options.level0_file_num_compaction_trigger = 5;
  options.max_background_compactions = 2;
  options.max_subcompactions = max_subcompactions_;
+  options.statistics = ROCKSDB_NAMESPACE::CreateDBStatistics();
+  options.write_buffer_size = 2 << 20;  // 2MB
+
+  BlockBasedTableOptions table_options;
+  table_options.block_cache = NewLRUCache(64 << 20);  // 64MB
+  table_options.cache_index_and_filter_blocks = true;
+  table_options.pin_l0_filter_and_index_blocks_in_cache = true;
+  options.table_factory.reset(new BlockBasedTableFactory(table_options));
+
  DestroyAndReopen(options);

  const size_t kValueSize = 1 << 20;
@@ -3214,6 +3223,7 @@ TEST_P(DBCompactionTestWithParam, IntraL0Compaction) {
      ASSERT_OK(Put(Key(i + 1), value));
    }
    ASSERT_OK(Flush());
+    ASSERT_EQ(i + 1, TestGetTickerCount(options, BLOCK_CACHE_INDEX_MISS));
  }
  dbfull()->TEST_WaitForCompact();
  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
@@ -3228,6 +3238,16 @@ TEST_P(DBCompactionTestWithParam, IntraL0Compaction) {
  for (int i = 0; i < 2; ++i) {
    ASSERT_GE(level_to_files[0][i].fd.file_size, 1 << 21);
  }
+
+  // The index/filter in the file produced by intra-L0 should not be pinned.
+  // That means clearing unref'd entries in block cache and re-accessing the
+  // file produced by intra-L0 should bump the index block miss count.
+  uint64_t prev_index_misses =
+      TestGetTickerCount(options, BLOCK_CACHE_INDEX_MISS);
+  table_options.block_cache->EraseUnRefEntries();
+  ASSERT_EQ("", Get(Key(0)));
+  ASSERT_EQ(prev_index_misses + 1,
+            TestGetTickerCount(options, BLOCK_CACHE_INDEX_MISS));
 }

 TEST_P(DBCompactionTestWithParam, IntraL0CompactionDoesNotObsoleteDeletions) {
@@ -1569,6 +1569,12 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
  auto sfm = static_cast<SstFileManagerImpl*>(
      impl->immutable_db_options_.sst_file_manager.get());
  if (s.ok() && sfm) {
+    // Set Statistics ptr for SstFileManager to dump the stats of
+    // DeleteScheduler.
+    sfm->SetStatisticsPtr(impl->immutable_db_options_.statistics);
+    ROCKS_LOG_INFO(impl->immutable_db_options_.info_log,
+                   "SstFileManager instance %p", sfm);
+
    // Notify SstFileManager about all sst files that already exist in
    // db_paths[0] and cf_paths[0] when the DB is opened.

@@ -376,6 +376,7 @@ TEST_F(DBSSTTest, RateLimitedDelete) {
  env_->SetTimeElapseOnlySleep(&options);
  options.disable_auto_compactions = true;
  options.env = env_;
+  options.statistics = CreateDBStatistics();

  int64_t rate_bytes_per_sec = 1024 * 10;  // 10 Kbs / Sec
  Status s;
@@ -425,6 +426,9 @@ TEST_F(DBSSTTest, RateLimitedDelete) {
  }
  ASSERT_GT(time_spent_deleting, expected_penlty * 0.9);
  ASSERT_LT(time_spent_deleting, expected_penlty * 1.1);
+  ASSERT_EQ(4, options.statistics->getAndResetTickerCount(FILES_MARKED_TRASH));
+  ASSERT_EQ(
+      0, options.statistics->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));

  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
 }
@@ -816,6 +820,12 @@ TEST_F(DBSSTTest, CancellingCompactionsWorks) {
  ASSERT_EQ(sfm->GetCompactionsReservedSize(), 0);
  // Make sure the stat is bumped
  ASSERT_GT(dbfull()->immutable_db_options().statistics.get()->getTickerCount(COMPACTION_CANCELLED), 0);
+  ASSERT_EQ(0,
+            dbfull()->immutable_db_options().statistics.get()->getTickerCount(
+                FILES_MARKED_TRASH));
+  ASSERT_EQ(4,
+            dbfull()->immutable_db_options().statistics.get()->getTickerCount(
+                FILES_DELETED_IMMEDIATELY));
  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
 }

@@ -25,6 +25,7 @@ class DBTest2 : public DBTestBase {
  DBTest2() : DBTestBase("/db_test2") {}
 };

+#ifndef ROCKSDB_LITE
 TEST_F(DBTest2, OpenForReadOnly) {
  DB* db_ptr = nullptr;
  std::string dbname = test::PerThreadDBPath("db_readonly");
@@ -85,6 +86,7 @@ TEST_F(DBTest2, OpenForReadOnlyWithColumnFamilies) {
  // With create_if_missing false, there should not be a dir in the file system
  ASSERT_NOK(env_->FileExists(dbname));
 }
+#endif  // ROCKSDB_LITE

 class PrefixFullBloomWithReverseComparator
    : public DBTestBase,
@@ -84,9 +84,9 @@ class ForwardLevelIterator : public InternalIterator {
        prefix_extractor_, /*table_reader_ptr=*/nullptr,
        /*file_read_hist=*/nullptr, TableReaderCaller::kUserIterator,
        /*arena=*/nullptr, /*skip_filters=*/false, /*level=*/-1,
+        /*max_file_size_for_l0_meta_pin=*/0,
        /*smallest_compaction_key=*/nullptr,
-        /*largest_compaction_key=*/nullptr,
-        allow_unprepared_value_);
+        /*largest_compaction_key=*/nullptr, allow_unprepared_value_);
    file_iter_->SetPinnedItersMgr(pinned_iters_mgr_);
    valid_ = false;
    if (!range_del_agg.IsEmpty()) {
@@ -686,9 +686,9 @@ void ForwardIterator::RebuildIterators(bool refresh_sv) {
        /*table_reader_ptr=*/nullptr, /*file_read_hist=*/nullptr,
        TableReaderCaller::kUserIterator, /*arena=*/nullptr,
        /*skip_filters=*/false, /*level=*/-1,
+        MaxFileSizeForL0MetaPin(sv_->mutable_cf_options),
        /*smallest_compaction_key=*/nullptr,
-        /*largest_compaction_key=*/nullptr,
-        allow_unprepared_value_));
+        /*largest_compaction_key=*/nullptr, allow_unprepared_value_));
  }
  BuildLevelIterators(vstorage);
  current_ = nullptr;
@@ -764,9 +764,9 @@ void ForwardIterator::RenewIterators() {
        /*table_reader_ptr=*/nullptr, /*file_read_hist=*/nullptr,
        TableReaderCaller::kUserIterator, /*arena=*/nullptr,
        /*skip_filters=*/false, /*level=*/-1,
+        MaxFileSizeForL0MetaPin(svnew->mutable_cf_options),
        /*smallest_compaction_key=*/nullptr,
-        /*largest_compaction_key=*/nullptr,
-        allow_unprepared_value_));
+        /*largest_compaction_key=*/nullptr, allow_unprepared_value_));
  }

  for (auto* f : l0_iters_) {
@@ -830,9 +830,9 @@ void ForwardIterator::ResetIncompleteIterators() {
        /*table_reader_ptr=*/nullptr, /*file_read_hist=*/nullptr,
        TableReaderCaller::kUserIterator, /*arena=*/nullptr,
        /*skip_filters=*/false, /*level=*/-1,
+        MaxFileSizeForL0MetaPin(sv_->mutable_cf_options),
        /*smallest_compaction_key=*/nullptr,
-        /*largest_compaction_key=*/nullptr,
-        allow_unprepared_value_);
+        /*largest_compaction_key=*/nullptr, allow_unprepared_value_);
    l0_iters_[i]->SetPinnedItersMgr(pinned_iters_mgr_);
  }

@@ -528,7 +528,8 @@ class Repairer {
          cfd->GetLatestMutableCFOptions()->prefix_extractor.get(),
          /*table_reader_ptr=*/nullptr, /*file_read_hist=*/nullptr,
          TableReaderCaller::kRepair, /*arena=*/nullptr, /*skip_filters=*/false,
-          /*level=*/-1, /*smallest_compaction_key=*/nullptr,
+          /*level=*/-1, /*max_file_size_for_l0_meta_pin=*/0,
+          /*smallest_compaction_key=*/nullptr,
          /*largest_compaction_key=*/nullptr,
          /*allow_unprepared_value=*/false);
      ParsedInternalKey parsed;
@@ -97,7 +97,8 @@ Status TableCache::GetTableReader(
    bool sequential_mode, bool record_read_stats, HistogramImpl* file_read_hist,
    std::unique_ptr<TableReader>* table_reader,
    const SliceTransform* prefix_extractor, bool skip_filters, int level,
-    bool prefetch_index_and_filter_in_cache) {
+    bool prefetch_index_and_filter_in_cache,
+    size_t max_file_size_for_l0_meta_pin) {
  std::string fname =
      TableFileName(ioptions_.cf_paths, fd.GetNumber(), fd.GetPathId());
  std::unique_ptr<FSRandomAccessFile> file;
@@ -124,7 +125,8 @@ Status TableCache::GetTableReader(
        TableReaderOptions(ioptions_, prefix_extractor, file_options,
                           internal_comparator, skip_filters, immortal_tables_,
                           false /* force_direct_prefetch */, level,
-                           fd.largest_seqno, block_cache_tracer_),
+                           fd.largest_seqno, block_cache_tracer_,
+                           max_file_size_for_l0_meta_pin),
        std::move(file_reader), fd.GetFileSize(), table_reader,
        prefetch_index_and_filter_in_cache);
    TEST_SYNC_POINT("TableCache::GetTableReader:0");
@@ -145,8 +147,8 @@ Status TableCache::FindTable(const FileOptions& file_options,
                             const SliceTransform* prefix_extractor,
                             const bool no_io, bool record_read_stats,
                             HistogramImpl* file_read_hist, bool skip_filters,
-                             int level,
-                             bool prefetch_index_and_filter_in_cache) {
+                             int level, bool prefetch_index_and_filter_in_cache,
+                             size_t max_file_size_for_l0_meta_pin) {
  PERF_TIMER_GUARD_WITH_ENV(find_table_nanos, ioptions_.env);
  Status s;
  uint64_t number = fd.GetNumber();
@@ -170,7 +172,8 @@ Status TableCache::FindTable(const FileOptions& file_options,
    s = GetTableReader(file_options, internal_comparator, fd,
                       false /* sequential mode */, record_read_stats,
                       file_read_hist, &table_reader, prefix_extractor,
-                       skip_filters, level, prefetch_index_and_filter_in_cache);
+                       skip_filters, level, prefetch_index_and_filter_in_cache,
+                       max_file_size_for_l0_meta_pin);
    if (!s.ok()) {
      assert(table_reader == nullptr);
      RecordTick(ioptions_.statistics, NO_FILE_ERRORS);
@@ -194,6 +197,7 @@ InternalIterator* TableCache::NewIterator(
    RangeDelAggregator* range_del_agg, const SliceTransform* prefix_extractor,
    TableReader** table_reader_ptr, HistogramImpl* file_read_hist,
    TableReaderCaller caller, Arena* arena, bool skip_filters, int level,
+    size_t max_file_size_for_l0_meta_pin,
    const InternalKey* smallest_compaction_key,
    const InternalKey* largest_compaction_key, bool allow_unprepared_value) {
  PERF_TIMER_GUARD(new_table_iterator_nanos);
@@ -211,7 +215,9 @@ InternalIterator* TableCache::NewIterator(
    s = FindTable(file_options, icomparator, fd, &handle, prefix_extractor,
                  options.read_tier == kBlockCacheTier /* no_io */,
                  !for_compaction /* record_read_stats */, file_read_hist,
-                  skip_filters, level);
+                  skip_filters, level,
+                  true /* prefetch_index_and_filter_in_cache */,
+                  max_file_size_for_l0_meta_pin);
    if (s.ok()) {
      table_reader = GetTableReaderFromHandle(handle);
    }
@@ -372,7 +378,7 @@ Status TableCache::Get(const ReadOptions& options,
                       GetContext* get_context,
                       const SliceTransform* prefix_extractor,
                       HistogramImpl* file_read_hist, bool skip_filters,
-                       int level) {
+                       int level, size_t max_file_size_for_l0_meta_pin) {
  auto& fd = file_meta.fd;
  std::string* row_cache_entry = nullptr;
  bool done = false;
@@ -397,10 +403,12 @@ Status TableCache::Get(const ReadOptions& options,
  Cache::Handle* handle = nullptr;
  if (!done && s.ok()) {
    if (t == nullptr) {
-      s = FindTable(
-          file_options_, internal_comparator, fd, &handle, prefix_extractor,
-          options.read_tier == kBlockCacheTier /* no_io */,
-          true /* record_read_stats */, file_read_hist, skip_filters, level);
+      s = FindTable(file_options_, internal_comparator, fd, &handle,
+                    prefix_extractor,
+                    options.read_tier == kBlockCacheTier /* no_io */,
+                    true /* record_read_stats */, file_read_hist, skip_filters,
+                    level, true /* prefetch_index_and_filter_in_cache */,
+                    max_file_size_for_l0_meta_pin);
      if (s.ok()) {
        t = GetTableReaderFromHandle(handle);
      }
@@ -72,7 +72,8 @@ class TableCache {
      const FileMetaData& file_meta, RangeDelAggregator* range_del_agg,
      const SliceTransform* prefix_extractor, TableReader** table_reader_ptr,
      HistogramImpl* file_read_hist, TableReaderCaller caller, Arena* arena,
-      bool skip_filters, int level, const InternalKey* smallest_compaction_key,
+      bool skip_filters, int level, size_t max_file_size_for_l0_meta_pin,
+      const InternalKey* smallest_compaction_key,
      const InternalKey* largest_compaction_key, bool allow_unprepared_value);

  // If a seek to internal key "k" in specified file finds an entry,
@@ -91,7 +92,7 @@ class TableCache {
             GetContext* get_context,
             const SliceTransform* prefix_extractor = nullptr,
             HistogramImpl* file_read_hist = nullptr, bool skip_filters = false,
-             int level = -1);
+             int level = -1, size_t max_file_size_for_l0_meta_pin = 0);

  // Return the range delete tombstone iterator of the file specified by
  // `file_meta`.
@@ -135,7 +136,8 @@ class TableCache {
                   const bool no_io = false, bool record_read_stats = true,
                   HistogramImpl* file_read_hist = nullptr,
                   bool skip_filters = false, int level = -1,
-                   bool prefetch_index_and_filter_in_cache = true);
+                   bool prefetch_index_and_filter_in_cache = true,
+                   size_t max_file_size_for_l0_meta_pin = 0);

  // Get TableReader from a cache handle.
  TableReader* GetTableReaderFromHandle(Cache::Handle* handle);
@@ -200,7 +202,8 @@ class TableCache {
                        std::unique_ptr<TableReader>* table_reader,
                        const SliceTransform* prefix_extractor = nullptr,
                        bool skip_filters = false, int level = -1,
-                        bool prefetch_index_and_filter_in_cache = true);
+                        bool prefetch_index_and_filter_in_cache = true,
+                        size_t max_file_size_for_l0_meta_pin = 0);

  // Create a key prefix for looking up the row cache. The prefix is of the
  // format row_cache_id + fd_number + seq_no. Later, the user key can be
@@ -278,7 +278,6 @@ class VersionBuilder::Rep {
          TEST_SYNC_POINT_CALLBACK("VersionBuilder::CheckConsistency0", &pair);
 #endif
          if (!level_zero_cmp_(f1, f2)) {
-            fprintf(stderr, "L0 files are not sorted properly");
            return Status::Corruption("L0 files are not sorted properly");
          }

@@ -287,11 +286,6 @@ class VersionBuilder::Rep {
            SequenceNumber external_file_seqno = f2->fd.smallest_seqno;
            if (!(external_file_seqno < f1->fd.largest_seqno ||
                  external_file_seqno == 0)) {
-              fprintf(stderr,
-                      "L0 file with seqno %" PRIu64 " %" PRIu64
-                      " vs. file with global_seqno %" PRIu64 "\n",
-                      f1->fd.smallest_seqno, f1->fd.largest_seqno,
-                      external_file_seqno);
              return Status::Corruption(
                  "L0 file with seqno " +
                  NumberToString(f1->fd.smallest_seqno) + " " +
@@ -301,11 +295,6 @@ class VersionBuilder::Rep {
                  NumberToString(f1->fd.GetNumber()));
            }
          } else if (f1->fd.smallest_seqno <= f2->fd.smallest_seqno) {
-            fprintf(stderr,
-                    "L0 files seqno %" PRIu64 " %" PRIu64 " vs. %" PRIu64
-                    " %" PRIu64 "\n",
-                    f1->fd.smallest_seqno, f1->fd.largest_seqno,
-                    f2->fd.smallest_seqno, f2->fd.largest_seqno);
            return Status::Corruption(
                "L0 files seqno " + NumberToString(f1->fd.smallest_seqno) +
                " " + NumberToString(f1->fd.largest_seqno) + " " +
@@ -320,7 +309,6 @@ class VersionBuilder::Rep {
          TEST_SYNC_POINT_CALLBACK("VersionBuilder::CheckConsistency1", &pair);
 #endif
          if (!level_nonzero_cmp_(f1, f2)) {
-            fprintf(stderr, "L%d files are not sorted properly", level);
            return Status::Corruption("L" + NumberToString(level) +
                                      " files are not sorted properly");
          }
@@ -328,9 +316,6 @@ class VersionBuilder::Rep {
          // Make sure there is no overlap in levels > 0
          if (vstorage->InternalComparator()->Compare(f1->largest,
                                                      f2->smallest) >= 0) {
-            fprintf(stderr, "L%d have overlapping ranges %s vs. %s\n", level,
-                    (f1->largest).DebugString(true).c_str(),
-                    (f2->smallest).DebugString(true).c_str());
            return Status::Corruption(
                "L" + NumberToString(level) + " have overlapping ranges " +
                (f1->largest).DebugString(true) + " vs. " +
@@ -769,7 +754,8 @@ class VersionBuilder::Rep {
  Status LoadTableHandlers(InternalStats* internal_stats, int max_threads,
                           bool prefetch_index_and_filter_in_cache,
                           bool is_initial_load,
-                           const SliceTransform* prefix_extractor) {
+                           const SliceTransform* prefix_extractor,
+                           size_t max_file_size_for_l0_meta_pin) {
    assert(table_cache_ != nullptr);

    size_t table_cache_capacity = table_cache_->get_cache()->GetCapacity();
@@ -838,7 +824,7 @@ class VersionBuilder::Rep {
            file_meta->fd, &file_meta->table_reader_handle, prefix_extractor,
            false /*no_io */, true /* record_read_stats */,
            internal_stats->GetFileReadHist(level), false, level,
-            prefetch_index_and_filter_in_cache);
+            prefetch_index_and_filter_in_cache, max_file_size_for_l0_meta_pin);
        if (file_meta->table_reader_handle != nullptr) {
          // Load table_reader
          file_meta->fd.table_reader = table_cache_->GetTableReaderFromHandle(
@@ -897,10 +883,11 @@ Status VersionBuilder::SaveTo(VersionStorageInfo* vstorage) {
 Status VersionBuilder::LoadTableHandlers(
    InternalStats* internal_stats, int max_threads,
    bool prefetch_index_and_filter_in_cache, bool is_initial_load,
-    const SliceTransform* prefix_extractor) {
-  return rep_->LoadTableHandlers(internal_stats, max_threads,
-                                 prefetch_index_and_filter_in_cache,
-                                 is_initial_load, prefix_extractor);
+    const SliceTransform* prefix_extractor,
+    size_t max_file_size_for_l0_meta_pin) {
+  return rep_->LoadTableHandlers(
+      internal_stats, max_threads, prefetch_index_and_filter_in_cache,
+      is_initial_load, prefix_extractor, max_file_size_for_l0_meta_pin);
 }

 BaseReferencedVersionBuilder::BaseReferencedVersionBuilder(
@@ -42,7 +42,8 @@ class VersionBuilder {
  Status LoadTableHandlers(InternalStats* internal_stats, int max_threads,
                           bool prefetch_index_and_filter_in_cache,
                           bool is_initial_load,
-                           const SliceTransform* prefix_extractor);
+                           const SliceTransform* prefix_extractor,
+                           size_t max_file_size_for_l0_meta_pin);

 private:
  class Rep;
@@ -422,8 +422,19 @@ TEST_F(VersionBuilderTest, ApplyFileDeletionAndAddition) {
  constexpr uint64_t file_number = 2345;
  constexpr char smallest[] = "bar";
  constexpr char largest[] = "foo";
+  constexpr uint64_t file_size = 10000;
+  constexpr uint32_t path_id = 0;
+  constexpr SequenceNumber smallest_seq = 100;
+  constexpr SequenceNumber largest_seq = 500;
+  constexpr uint64_t num_entries = 0;
+  constexpr uint64_t num_deletions = 0;
+  constexpr bool sampled = false;
+  constexpr SequenceNumber smallest_seqno = 1;
+  constexpr SequenceNumber largest_seqno = 1000;

-  Add(level, file_number, smallest, largest);
+  Add(level, file_number, smallest, largest, file_size, path_id, smallest_seq,
+      largest_seq, num_entries, num_deletions, sampled, smallest_seqno,
+      largest_seqno);

  EnvOptions env_options;
  constexpr TableCache* table_cache = nullptr;
@@ -440,18 +451,14 @@ TEST_F(VersionBuilderTest, ApplyFileDeletionAndAddition) {

  VersionEdit addition;

-  constexpr uint32_t path_id = 0;
-  constexpr uint64_t file_size = 10000;
-  constexpr SequenceNumber smallest_seqno = 100;
-  constexpr SequenceNumber largest_seqno = 1000;
  constexpr bool marked_for_compaction = false;

  addition.AddFile(level, file_number, path_id, file_size,
-                   GetInternalKey(smallest), GetInternalKey(largest),
-                   smallest_seqno, largest_seqno, marked_for_compaction,
-                   kInvalidBlobFileNumber, kUnknownOldestAncesterTime,
-                   kUnknownFileCreationTime, kUnknownFileChecksum,
-                   kUnknownFileChecksumFuncName);
+                   GetInternalKey(smallest, smallest_seq),
+                   GetInternalKey(largest, largest_seq), smallest_seqno,
+                   largest_seqno, marked_for_compaction, kInvalidBlobFileNumber,
+                   kUnknownOldestAncesterTime, kUnknownFileCreationTime,
+                   kUnknownFileChecksum, kUnknownFileChecksumFuncName);

  ASSERT_OK(builder.Apply(&addition));

@@ -18,10 +18,6 @@
 #include "util/string_util.h"

 namespace ROCKSDB_NAMESPACE {
-// The unknown file checksum.
-const std::string kUnknownFileChecksum("");
-// The unknown sst file checksum function name.
-const std::string kUnknownFileChecksumFuncName("Unknown");

 namespace {

@@ -29,9 +29,6 @@ constexpr uint64_t kFileNumberMask = 0x3FFFFFFFFFFFFFFF;
 constexpr uint64_t kUnknownOldestAncesterTime = 0;
 constexpr uint64_t kUnknownFileCreationTime = 0;

-extern const std::string kUnknownFileChecksum;
-extern const std::string kUnknownFileChecksumFuncName;
-
 extern uint64_t PackFileNumberAndPathId(uint64_t number, uint64_t path_id);

 // A copyable structure contains information needed to read data from an SST
@@ -414,7 +414,8 @@ Status VersionEditHandler::LoadTables(ColumnFamilyData* cfd,
      cfd->internal_stats(),
      version_set_->db_options_->max_file_opening_threads,
      prefetch_index_and_filter_in_cache, is_initial_load,
-      cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
+      cfd->GetLatestMutableCFOptions()->prefix_extractor.get(),
+      MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
  if ((s.IsPathNotFound() || s.IsCorruption()) &&
      no_error_if_table_files_missing_) {
    s = Status::OK();
@@ -365,6 +365,7 @@ class FilePickerMultiGet {
        range_(range),
        batch_iter_(range->begin()),
        batch_iter_prev_(range->begin()),
+        upper_key_(range->begin()),
        maybe_repeat_key_(false),
        current_level_range_(*range, range->begin(), range->end()),
        current_file_range_(*range, range->begin(), range->end()),
@@ -481,9 +482,19 @@ class FilePickerMultiGet {
      }
      if (cmp_largest == 0) {
        // cmp_largest is 0, which means the next key will not be in this
-        // file, so stop looking further. Also don't increment megt_iter_
-        // as we may have to look for this key in the next file if we don't
-        // find it in this one
+        // file, so stop looking further. However, its possible there are
+        // duplicates in the batch, so find the upper bound for the batch
+        // in this file (upper_key_) by skipping past the duplicates. We
+        // leave batch_iter_ as is since we may have to pick up from there
+        // for the next file, if this file has a merge value rather than
+        // final value
+        upper_key_ = batch_iter_;
+        ++upper_key_;
+        while (upper_key_ != current_level_range_.end() &&
+               user_comparator_->Compare(batch_iter_->ukey, upper_key_->ukey) ==
+                   0) {
+          ++upper_key_;
+        }
        break;
      } else {
        if (curr_level_ == 0) {
@@ -503,6 +514,12 @@ class FilePickerMultiGet {
    *fd = f;
    *file_index = curr_file_index;
    *is_last_key_in_file = cmp_largest == 0;
+    if (!*is_last_key_in_file) {
+      // If the largest key in the batch overlapping the file is not the
+      // largest key in the file, upper_ley_ would not have been updated so
+      // update it here
+      upper_key_ = batch_iter_;
+    }
    return file_hit;
  }

@@ -524,7 +541,7 @@ class FilePickerMultiGet {
          // file regardless for all keys not found yet
          if (current_level_range_.CheckKeyDone(batch_iter_) ||
              curr_level_ == 0) {
-            ++batch_iter_;
+            batch_iter_ = upper_key_;
          }
        }
        // batch_iter_prev_ will become the start key for the next file
@@ -544,18 +561,20 @@ class FilePickerMultiGet {
                                      &is_last_key_in_file)) {
        search_ended_ = !PrepareNextLevel();
      } else {
-        MultiGetRange::Iterator upper_key = batch_iter_;
        if (is_last_key_in_file) {
          // Since cmp_largest is 0, batch_iter_ still points to the last key
          // that falls in this file, instead of the next one. Increment
-          // upper_key so we can set the range properly for SST MultiGet
-          ++upper_key;
-          ++(fp_ctx_array_[batch_iter_.index()].curr_index_in_curr_level);
+          // the file index for all keys between batch_iter_ and upper_key_
+          auto tmp_iter = batch_iter_;
+          while (tmp_iter != upper_key_) {
+            ++(fp_ctx_array_[tmp_iter.index()].curr_index_in_curr_level);
+            ++tmp_iter;
+          }
          maybe_repeat_key_ = true;
        }
        // Set the range for this file
        current_file_range_ =
-            MultiGetRange(next_file_range, batch_iter_prev_, upper_key);
+            MultiGetRange(next_file_range, batch_iter_prev_, upper_key_);
        returned_file_level_ = curr_level_;
        hit_file_level_ = curr_level_;
        is_hit_file_last_in_level_ =
@@ -607,6 +626,7 @@ class FilePickerMultiGet {
  // key found in the previous SST file, in order to serve as the start of
  // the batch key range for the next SST file
  MultiGetRange::Iterator batch_iter_prev_;
+  MultiGetRange::Iterator upper_key_;
  bool maybe_repeat_key_;
  MultiGetRange current_level_range_;
  MultiGetRange current_file_range_;
@@ -626,7 +646,7 @@ class FilePickerMultiGet {
      if (fp_ctx_array_[mget_iter.index()].curr_index_in_curr_level <
          curr_file_level_->num_files) {
        batch_iter_prev_ = current_level_range_.begin();
-        batch_iter_ = current_level_range_.begin();
+        upper_key_ = batch_iter_ = current_level_range_.begin();
        return true;
      }
    }
@@ -721,7 +741,7 @@ class FilePickerMultiGet {
      }
      if (level_contains_keys) {
        batch_iter_prev_ = current_level_range_.begin();
-        batch_iter_ = current_level_range_.begin();
+        upper_key_ = batch_iter_ = current_level_range_.begin();
        return true;
      }
      curr_level_++;
@@ -984,7 +1004,8 @@ class LevelIterator final : public InternalIterator {
        read_options_, file_options_, icomparator_, *file_meta.file_metadata,
        range_del_agg_, prefix_extractor_,
        nullptr /* don't need reference to table */, file_read_hist_, caller_,
-        /*arena=*/nullptr, skip_filters_, level_, smallest_compaction_key,
+        /*arena=*/nullptr, skip_filters_, level_,
+        /*max_file_size_for_l0_meta_pin=*/0, smallest_compaction_key,
        largest_compaction_key, allow_unprepared_value_);
  }

@@ -1579,10 +1600,9 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options,
          mutable_cf_options_.prefix_extractor.get(), nullptr,
          cfd_->internal_stats()->GetFileReadHist(0),
          TableReaderCaller::kUserIterator, arena,
-          /*skip_filters=*/false, /*level=*/0,
+          /*skip_filters=*/false, /*level=*/0, max_file_size_for_l0_meta_pin_,
          /*smallest_compaction_key=*/nullptr,
-          /*largest_compaction_key=*/nullptr,
-          allow_unprepared_value));
+          /*largest_compaction_key=*/nullptr, allow_unprepared_value));
    }
    if (should_sample) {
      // Count ones for every L0 files. This is done per iterator creation
@@ -1604,8 +1624,8 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options,
        mutable_cf_options_.prefix_extractor.get(), should_sample_file_read(),
        cfd_->internal_stats()->GetFileReadHist(level),
        TableReaderCaller::kUserIterator, IsFilterSkipped(level), level,
-        range_del_agg, /*largest_compaction_key=*/nullptr,
-        allow_unprepared_value));
+        range_del_agg,
+        /*compaction_boundaries=*/nullptr, allow_unprepared_value));
  }
 }

@@ -1639,7 +1659,7 @@ Status Version::OverlapWithLevelIterator(const ReadOptions& read_options,
          mutable_cf_options_.prefix_extractor.get(), nullptr,
          cfd_->internal_stats()->GetFileReadHist(0),
          TableReaderCaller::kUserIterator, &arena,
-          /*skip_filters=*/false, /*level=*/0,
+          /*skip_filters=*/false, /*level=*/0, max_file_size_for_l0_meta_pin_,
          /*smallest_compaction_key=*/nullptr,
          /*largest_compaction_key=*/nullptr,
          /*allow_unprepared_value=*/false));
@@ -1743,6 +1763,8 @@ Version::Version(ColumnFamilyData* column_family_data, VersionSet* vset,
      refs_(0),
      file_options_(file_opt),
      mutable_cf_options_(mutable_cf_options),
+      max_file_size_for_l0_meta_pin_(
+          MaxFileSizeForL0MetaPin(mutable_cf_options_)),
      version_number_(version_number) {}

 void Version::Get(const ReadOptions& read_options, const LookupKey& k,
@@ -1806,7 +1828,7 @@ void Version::Get(const ReadOptions& read_options, const LookupKey& k,
        cfd_->internal_stats()->GetFileReadHist(fp.GetHitFileLevel()),
        IsFilterSkipped(static_cast<int>(fp.GetHitFileLevel()),
                        fp.IsHitFileLastInLevel()),
-        fp.GetHitFileLevel());
+        fp.GetHitFileLevel(), max_file_size_for_l0_meta_pin_);
    // TODO: examine the behavior for corrupted key
    if (timer_enabled) {
      PERF_COUNTER_BY_LEVEL_ADD(get_from_table_nanos, timer.ElapsedNanos(),
@@ -3902,7 +3924,8 @@ Status VersionSet::ProcessManifestWrites(
            cfd->internal_stats(), 1 /* max_threads */,
            true /* prefetch_index_and_filter_in_cache */,
            false /* is_initial_load */,
-            mutable_cf_options_ptrs[i]->prefix_extractor.get());
+            mutable_cf_options_ptrs[i]->prefix_extractor.get(),
+            MaxFileSizeForL0MetaPin(*mutable_cf_options_ptrs[i]));
        if (!s.ok()) {
          if (db_options_->paranoid_checks) {
            break;
@@ -4629,7 +4652,8 @@ Status VersionSet::Recover(
          cfd->internal_stats(), db_options_->max_file_opening_threads,
          false /* prefetch_index_and_filter_in_cache */,
          true /* is_initial_load */,
-          cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
+          cfd->GetLatestMutableCFOptions()->prefix_extractor.get(),
+          MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
      if (!s.ok()) {
        if (db_options_->paranoid_checks) {
          return s;
@@ -5666,13 +5690,14 @@ InternalIterator* VersionSet::MakeInputIterator(
        for (size_t i = 0; i < flevel->num_files; i++) {
          list[num++] = cfd->table_cache()->NewIterator(
              read_options, file_options_compactions,
-              cfd->internal_comparator(),
-              *flevel->files[i].file_metadata, range_del_agg,
-              c->mutable_cf_options()->prefix_extractor.get(),
+              cfd->internal_comparator(), *flevel->files[i].file_metadata,
+              range_del_agg, c->mutable_cf_options()->prefix_extractor.get(),
              /*table_reader_ptr=*/nullptr,
              /*file_read_hist=*/nullptr, TableReaderCaller::kCompaction,
              /*arena=*/nullptr,
-              /*skip_filters=*/false, /*level=*/static_cast<int>(c->level(which)),
+              /*skip_filters=*/false,
+              /*level=*/static_cast<int>(c->level(which)),
+              MaxFileSizeForL0MetaPin(*c->mutable_cf_options()),
              /*smallest_compaction_key=*/nullptr,
              /*largest_compaction_key=*/nullptr,
              /*allow_unprepared_value=*/false);
@@ -6003,7 +6028,8 @@ Status ReactiveVersionSet::Recover(
                cfd->internal_stats(), db_options_->max_file_opening_threads,
                false /* prefetch_index_and_filter_in_cache */,
                true /* is_initial_load */,
-                cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
+                cfd->GetLatestMutableCFOptions()->prefix_extractor.get(),
+                MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
            if (!s.ok()) {
              enough = false;
              if (s.IsPathNotFound()) {
@@ -6279,7 +6305,8 @@ Status ReactiveVersionSet::ApplyOneVersionEditToBuilder(
        cfd->internal_stats(), db_options_->max_file_opening_threads,
        false /* prefetch_index_and_filter_in_cache */,
        false /* is_initial_load */,
-        cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
+        cfd->GetLatestMutableCFOptions()->prefix_extractor.get(),
+        MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
    TEST_SYNC_POINT_CALLBACK(
        "ReactiveVersionSet::ApplyOneVersionEditToBuilder:"
        "AfterLoadTableHandlers",
@@ -324,6 +324,17 @@ class VersionStorageInfo {
    return it->second;
  }

+  // REQUIRES: This version has been saved (see VersionSet::SaveTo)
+  FileMetaData* GetFileMetaDataByNumber(uint64_t file_number) const {
+    auto location = GetFileLocation(file_number);
+
+    if (!location.IsValid()) {
+      return nullptr;
+    }
+
+    return files_[location.GetLevel()][location.GetPosition()];
+  }
+
  // REQUIRES: This version has been saved (see VersionSet::SaveTo)
  using BlobFiles = std::map<uint64_t, std::shared_ptr<BlobFileMetaData>>;
  const BlobFiles& GetBlobFiles() const { return blob_files_; }
@@ -792,6 +803,8 @@ class Version {
  int refs_;                    // Number of live refs to this version
  const FileOptions file_options_;
  const MutableCFOptions mutable_cf_options_;
+  // Cached value to avoid recomputing it on every read.
+  const size_t max_file_size_for_l0_meta_pin_;

  // A version number that uniquely represents this version. This is
  // used for debugging and logging purposes only.
@@ -421,7 +421,7 @@ TEST_F(VersionStorageInfoTest, GetOverlappingInputs) {
      1, {"i", 0, kTypeValue}, {"j", 0, kTypeValue}));
 }

-TEST_F(VersionStorageInfoTest, FileLocation) {
+TEST_F(VersionStorageInfoTest, FileLocationAndMetaDataByNumber) {
  Add(0, 11U, "1", "2", 5000U);
  Add(0, 12U, "1", "2", 5000U);

@@ -429,13 +429,18 @@ TEST_F(VersionStorageInfoTest, FileLocation) {

  ASSERT_EQ(vstorage_.GetFileLocation(11U),
            VersionStorageInfo::FileLocation(0, 0));
+  ASSERT_NE(vstorage_.GetFileMetaDataByNumber(11U), nullptr);
+
  ASSERT_EQ(vstorage_.GetFileLocation(12U),
            VersionStorageInfo::FileLocation(0, 1));
+  ASSERT_NE(vstorage_.GetFileMetaDataByNumber(12U), nullptr);

  ASSERT_EQ(vstorage_.GetFileLocation(7U),
            VersionStorageInfo::FileLocation(2, 0));
+  ASSERT_NE(vstorage_.GetFileMetaDataByNumber(7U), nullptr);

  ASSERT_FALSE(vstorage_.GetFileLocation(999U).IsValid());
+  ASSERT_EQ(vstorage_.GetFileMetaDataByNumber(999U), nullptr);
 }

 class VersionStorageInfoTimestampTest : public VersionStorageInfoTestBase {
@@ -7,6 +7,7 @@

 #include "file/delete_scheduler.h"

+#include <cinttypes>
 #include <thread>
 #include <vector>

@@ -65,6 +66,12 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
    s = fs_->DeleteFile(file_path, IOOptions(), nullptr);
    if (s.ok()) {
      sst_file_manager_->OnDeleteFile(file_path);
+      ROCKS_LOG_INFO(info_log_,
+                     "Deleted file %s immediately, rate_bytes_per_sec %" PRIi64
+                     ", total_trash_size %" PRIu64 " max_trash_db_ratio %lf",
+                     file_path.c_str(), rate_bytes_per_sec_.load(),
+                     total_trash_size_.load(), max_trash_db_ratio_.load());
+      RecordTick(stats_.get(), FILES_DELETED_IMMEDIATELY);
    }
    return s;
  }
@@ -72,6 +79,8 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
  // Move file to trash
  std::string trash_file;
  s = MarkAsTrash(file_path, &trash_file);
+  ROCKS_LOG_INFO(info_log_, "Mark file: %s as trash -- %s", trash_file.c_str(),
+                 s.ToString().c_str());

  if (!s.ok()) {
    ROCKS_LOG_ERROR(info_log_, "Failed to mark %s as trash -- %s",
@@ -79,10 +88,14 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
    s = fs_->DeleteFile(file_path, IOOptions(), nullptr);
    if (s.ok()) {
      sst_file_manager_->OnDeleteFile(file_path);
+      ROCKS_LOG_INFO(info_log_, "Deleted file %s immediately",
+                     trash_file.c_str());
+      RecordTick(stats_.get(), FILES_DELETED_IMMEDIATELY);
    }
    return s;
  }

+  RecordTick(stats_.get(), FILES_MARKED_TRASH);
  // Update the total trash size
  uint64_t trash_file_size = 0;
  fs_->GetFileSize(trash_file, IOOptions(), &trash_file_size, nullptr);
@@ -210,6 +223,8 @@ void DeleteScheduler::BackgroundEmptyTrash() {
        current_delete_rate = rate_bytes_per_sec_.load();
        start_time = env_->NowMicros();
        total_deleted_bytes = 0;
+        ROCKS_LOG_INFO(info_log_, "rate_bytes_per_sec is changed to %" PRIi64,
+                       current_delete_rate);
      }

      // Get new file to delete
@@ -233,19 +248,27 @@ void DeleteScheduler::BackgroundEmptyTrash() {
        bg_errors_[path_in_trash] = s;
      }

-      // Apply penlty if necessary
-      uint64_t total_penlty;
+      // Apply penalty if necessary
+      uint64_t total_penalty;
      if (current_delete_rate > 0) {
        // rate limiting is enabled
-        total_penlty =
+        total_penalty =
            ((total_deleted_bytes * kMicrosInSecond) / current_delete_rate);
-        while (!closing_ && !cv_.TimedWait(start_time + total_penlty)) {}
+        ROCKS_LOG_INFO(info_log_,
+                       "Rate limiting is enabled with penalty %" PRIu64
+                       "after deleting file %s",
+                       total_penalty, path_in_trash.c_str());
+        while (!closing_ && !cv_.TimedWait(start_time + total_penalty)) {
+        }
      } else {
        // rate limiting is disabled
-        total_penlty = 0;
+        total_penalty = 0;
+        ROCKS_LOG_INFO(info_log_,
+                       "Rate limiting is disabled after deleting file %s",
+                       path_in_trash.c_str());
      }
      TEST_SYNC_POINT_CALLBACK("DeleteScheduler::BackgroundEmptyTrash:Wait",
-                               &total_penlty);
+                               &total_penalty);

      if (is_complete) {
        pending_files_--;
@@ -353,9 +376,13 @@ void DeleteScheduler::WaitForEmptyTrash() {
 }

 void DeleteScheduler::MaybeCreateBackgroundThread() {
-  if(bg_thread_ == nullptr && rate_bytes_per_sec_.load() > 0) {
+  if (bg_thread_ == nullptr && rate_bytes_per_sec_.load() > 0) {
    bg_thread_.reset(
        new port::Thread(&DeleteScheduler::BackgroundEmptyTrash, this));
+    ROCKS_LOG_INFO(info_log_,
+                   "Created background thread for deletion scheduler with "
+                   "rate_bytes_per_sec: %" PRIi64,
+                   rate_bytes_per_sec_.load());
  }
 }

@@ -83,6 +83,10 @@ class DeleteScheduler {
  static Status CleanupDirectory(Env* env, SstFileManagerImpl* sfm,
                                 const std::string& path);

+  void SetStatisticsPtr(const std::shared_ptr<Statistics>& stats) {
+    stats_ = stats;
+  }
+
 private:
  Status MarkAsTrash(const std::string& file_path, std::string* path_in_trash);

@@ -137,6 +141,7 @@ class DeleteScheduler {
  // immediately
  std::atomic<double> max_trash_db_ratio_;
  static const uint64_t kMicrosInSecond = 1000 * 1000LL;
+  std::shared_ptr<Statistics> stats_;
 };

 }  // namespace ROCKSDB_NAMESPACE
@@ -32,6 +32,7 @@ class DeleteSchedulerTest : public testing::Test {
          ToString(i));
      DestroyAndCreateDir(dummy_files_dirs_.back());
    }
+    stats_ = ROCKSDB_NAMESPACE::CreateDBStatistics();
  }

  ~DeleteSchedulerTest() override {
@@ -99,6 +100,7 @@ class DeleteSchedulerTest : public testing::Test {
        new SstFileManagerImpl(env_, fs, nullptr, rate_bytes_per_sec_,
                               /* max_trash_db_ratio= */ 1.1, 128 * 1024));
    delete_scheduler_ = sst_file_mgr_->delete_scheduler();
+    sst_file_mgr_->SetStatisticsPtr(stats_);
  }

  Env* env_;
@@ -106,6 +108,7 @@ class DeleteSchedulerTest : public testing::Test {
  int64_t rate_bytes_per_sec_;
  DeleteScheduler* delete_scheduler_;
  std::unique_ptr<SstFileManagerImpl> sst_file_mgr_;
+  std::shared_ptr<Statistics> stats_;
 };

 // Test the basic functionality of DeleteScheduler (Rate Limiting).
@@ -182,6 +185,8 @@ TEST_F(DeleteSchedulerTest, BasicRateLimiting) {
    ASSERT_EQ(num_files, dir_synced);

    ASSERT_EQ(CountTrashFiles(), 0);
+    ASSERT_EQ(num_files, stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+    ASSERT_EQ(0, stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));
    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
  }
 }
@@ -219,6 +224,9 @@ TEST_F(DeleteSchedulerTest, MultiDirectoryDeletionsScheduled) {
    ASSERT_EQ(0, CountTrashFiles(i));
  }

+  ASSERT_EQ(kNumFiles, stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+  ASSERT_EQ(0, stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));
+
  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
 }

@@ -301,6 +309,10 @@ TEST_F(DeleteSchedulerTest, RateLimitingMultiThreaded) {

    ASSERT_EQ(CountNormalFiles(), 0);
    ASSERT_EQ(CountTrashFiles(), 0);
+    ASSERT_EQ(num_files * thread_cnt,
+              stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+    ASSERT_EQ(0, stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));
+
    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
  }
 }
@@ -318,8 +330,9 @@ TEST_F(DeleteSchedulerTest, DisableRateLimiting) {

  rate_bytes_per_sec_ = 0;
  NewDeleteScheduler();
+  constexpr int num_files = 10;

-  for (int i = 0; i < 10; i++) {
+  for (int i = 0; i < num_files; i++) {
    // Every file we delete will be deleted immediately
    std::string dummy_file = NewDummyFile("dummy.data");
    ASSERT_OK(delete_scheduler_->DeleteFile(dummy_file, ""));
@@ -329,6 +342,9 @@ TEST_F(DeleteSchedulerTest, DisableRateLimiting) {
  }

  ASSERT_EQ(bg_delete_file, 0);
+  ASSERT_EQ(0, stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+  ASSERT_EQ(num_files,
+            stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));

  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
 }
@@ -365,6 +381,8 @@ TEST_F(DeleteSchedulerTest, ConflictNames) {

  auto bg_errors = delete_scheduler_->GetBackgroundErrors();
  ASSERT_EQ(bg_errors.size(), 0);
+  ASSERT_EQ(10, stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+  ASSERT_EQ(0, stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));

  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
 }
@@ -439,9 +457,12 @@ TEST_F(DeleteSchedulerTest, StartBGEmptyTrashMultipleTimes) {

    auto bg_errors = delete_scheduler_->GetBackgroundErrors();
    ASSERT_EQ(bg_errors.size(), 0);
+    ASSERT_EQ(10, stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+    ASSERT_EQ(0, stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));
  }

  ASSERT_EQ(bg_delete_file, 50);
+
  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
 }

@@ -653,6 +674,8 @@ TEST_F(DeleteSchedulerTest, ImmediateDeleteOn25PercDBSize) {
  // When we end up with 26 files in trash we will start
  // deleting new files immediately
  ASSERT_EQ(fg_delete_file, 74);
+  ASSERT_EQ(26, stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+  ASSERT_EQ(74, stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));

  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
 }
@@ -135,6 +135,11 @@ class SstFileManagerImpl : public SstFileManager {
  // once in the object's lifetime, and before the destructor
  void Close();

+  void SetStatisticsPtr(const std::shared_ptr<Statistics>& stats) override {
+    stats_ = stats;
+    delete_scheduler_.SetStatisticsPtr(stats);
+  }
+
 private:
  // REQUIRES: mutex locked
  void OnAddFileImpl(const std::string& file_path, uint64_t file_size,
@@ -190,6 +195,7 @@ class SstFileManagerImpl : public SstFileManager {
  std::list<ErrorHandler*> error_handler_list_;
  // Pointer to ErrorHandler instance that is currently processing recovery
  ErrorHandler* cur_instance_;
+  std::shared_ptr<Statistics> stats_;
 };

 }  // namespace ROCKSDB_NAMESPACE
@@ -236,7 +236,7 @@ const char* WritableFileWriter::GetFileChecksumFuncName() const {
  if (checksum_generator_ != nullptr) {
    return checksum_generator_->Name();
  } else {
-    return kUnknownFileChecksumFuncName.c_str();
+    return kUnknownFileChecksumFuncName;
  }
 }

@@ -18,6 +18,11 @@

 namespace ROCKSDB_NAMESPACE {

+// The unknown file checksum.
+constexpr char kUnknownFileChecksum[] = "";
+// The unknown sst file checksum function name.
+constexpr char kUnknownFileChecksumFuncName[] = "Unknown";
+
 struct FileChecksumGenContext {
  std::string file_name;
 };
@@ -11,7 +11,7 @@ namespace ROCKSDB_NAMESPACE {

 class SSTDumpTool {
 public:
-  int Run(int argc, char** argv, Options options = Options());
+  int Run(int argc, char const* const* argv, Options options = Options());
 };

 }  // namespace ROCKSDB_NAMESPACE
@@ -11,6 +11,7 @@
 #include <vector>

 #include "rocksdb/file_system.h"
+#include "rocksdb/statistics.h"
 #include "rocksdb/status.h"

 namespace ROCKSDB_NAMESPACE {
@@ -80,6 +81,9 @@ class SstFileManager {
  // Return the total size of trash files
  // thread-safe
  virtual uint64_t GetTotalTrashSize() = 0;
+
+  // Set the statistics ptr to dump the stat information
+  virtual void SetStatisticsPtr(const std::shared_ptr<Statistics>& stats) = 0;
 };

 // Create a new SstFileManager that can be shared among multiple RocksDB
@@ -360,6 +360,12 @@ enum Tickers : uint32_t {
  //           <= BLOCK_CACHE_COMPRESSION_DICT_ADD
  BLOCK_CACHE_COMPRESSION_DICT_ADD_REDUNDANT,

+  // # of files marked as trash by sst file manager and will be deleted
+  // later by background thread.
+  FILES_MARKED_TRASH,
+  // # of files deleted immediately by sst file manger through delete scheduler.
+  FILES_DELETED_IMMEDIATELY,
+
  TICKER_ENUM_MAX
 };

@@ -78,7 +78,7 @@ class LDBCommand {
          SelectCommand);

  static LDBCommand* InitFromCmdLineArgs(
-      int argc, char** argv, const Options& options,
+      int argc, char const* const* argv, const Options& options,
      const LDBOptions& ldb_options,
      const std::vector<ColumnFamilyDescriptor>* column_families);

@@ -269,11 +269,13 @@ class LDBCommand {

 class LDBCommandRunner {
 public:
-  static void PrintHelp(const LDBOptions& ldb_options, const char* exec_name);
+  static void PrintHelp(const LDBOptions& ldb_options, const char* exec_name,
+                        bool to_stderr = true);

  // Returns the status code to return. 0 is no error.
  static int RunCommand(
-      int argc, char** argv, Options options, const LDBOptions& ldb_options,
+      int argc, char const* const* argv, Options options,
+      const LDBOptions& ldb_options,
      const std::vector<ColumnFamilyDescriptor>* column_families);
 };

@@ -1292,9 +1292,9 @@ class ByteBufferJni : public JavaClass {
    return constructWith(env, direct, nullptr, capacity, jbytebuffer_clazz);
  }

-  static jobject constructWith(
-      JNIEnv* env, const bool direct, const char* buf, const size_t capacity, 
-      jclass jbytebuffer_clazz = nullptr) {
+  static jobject constructWith(JNIEnv* env, const bool direct, const char* buf,
+                               const size_t capacity,
+                               jclass jbytebuffer_clazz = nullptr) {
    if (direct) {
      bool allocated = false;
      if (buf == nullptr) {
@@ -4945,6 +4945,11 @@ class TickerTypeJni {
        return -0x0C;
      case ROCKSDB_NAMESPACE::Tickers::TXN_GET_TRY_AGAIN:
        return -0x0D;
+      case ROCKSDB_NAMESPACE::Tickers::FILES_MARKED_TRASH:
+        return -0x0E;
+      case ROCKSDB_NAMESPACE::Tickers::FILES_DELETED_IMMEDIATELY:
+        return -0X0F;
+
      case ROCKSDB_NAMESPACE::Tickers::TICKER_ENUM_MAX:
        // 0x5F for backwards compatibility on current minor version.
        return 0x5F;
@@ -5240,6 +5245,10 @@ class TickerTypeJni {
        return ROCKSDB_NAMESPACE::Tickers::TXN_SNAPSHOT_MUTEX_OVERHEAD;
      case -0x0D:
        return ROCKSDB_NAMESPACE::Tickers::TXN_GET_TRY_AGAIN;
+      case -0x0E:
+        return ROCKSDB_NAMESPACE::Tickers::FILES_MARKED_TRASH;
+      case -0x0F:
+        return ROCKSDB_NAMESPACE::Tickers::FILES_DELETED_IMMEDIATELY;
      case 0x5F:
        // 0x5F for backwards compatibility on current minor version.
        return ROCKSDB_NAMESPACE::Tickers::TICKER_ENUM_MAX;
@@ -722,6 +722,16 @@ public enum TickerType {
     */
    TXN_GET_TRY_AGAIN((byte) -0x0D),

+    /**
+     * # of files marked as trash by delete scheduler
+     */
+    FILES_MARKED_TRASH((byte) -0x0E),
+
+    /**
+     * # of files deleted immediately by delete scheduler
+     */
+    FILES_DELETED_IMMEDIATELY((byte) -0x0f),
+
    TICKER_ENUM_MAX((byte) 0x5F);

    private final byte value;
@@ -184,6 +184,8 @@ const std::vector<std::pair<Tickers, std::string>> TickersNameMap = {
    {BLOCK_CACHE_DATA_ADD_REDUNDANT, "rocksdb.block.cache.data.add.redundant"},
    {BLOCK_CACHE_COMPRESSION_DICT_ADD_REDUNDANT,
     "rocksdb.block.cache.compression.dict.add.redundant"},
+    {FILES_MARKED_TRASH, "rocksdb.files.marked.trash"},
+    {FILES_DELETED_IMMEDIATELY, "rocksdb.files.deleted.immediately"},
 };

 const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {
@@ -720,6 +720,17 @@ uint64_t MaxFileSizeForLevel(const MutableCFOptions& cf_options,
  }
 }

+size_t MaxFileSizeForL0MetaPin(const MutableCFOptions& cf_options) {
+  // We do not want to pin meta-blocks that almost certainly came from intra-L0
+  // or a former larger `write_buffer_size` value to avoid surprising users with
+  // pinned memory usage. We use a factor of 1.5 to account for overhead
+  // introduced during flush in most cases.
+  if (port::kMaxSizet / 3 < cf_options.write_buffer_size / 2) {
+    return port::kMaxSizet;
+  }
+  return cf_options.write_buffer_size / 2 * 3;
+}
+
 void MutableCFOptions::RefreshDerivedOptions(int num_levels,
                                             CompactionStyle compaction_style) {
  max_file_size.resize(num_levels);
@@ -268,4 +268,9 @@ uint64_t MultiplyCheckOverflow(uint64_t op1, double op2);
 uint64_t MaxFileSizeForLevel(const MutableCFOptions& cf_options,
    int level, CompactionStyle compaction_style, int base_level = 1,
    bool level_compaction_dynamic_level_bytes = false);
+
+// Get the max size of an L0 file for which we will pin its meta-blocks when
+// `pin_l0_filter_and_index_blocks_in_cache` is set.
+size_t MaxFileSizeForL0MetaPin(const MutableCFOptions& cf_options);
+
 }  // namespace ROCKSDB_NAMESPACE
@@ -7,7 +7,6 @@

 #include <cinttypes>

-#include "db/version_edit.h"
 #include "logging/logging.h"
 #include "options/options_helper.h"
 #include "port/port.h"
@@ -616,9 +615,8 @@ void ImmutableDBOptions::Dump(Logger* log) const {
      log, "                Options.log_readahead_size: %" ROCKSDB_PRIszt,
      log_readahead_size);
  ROCKS_LOG_HEADER(log, "                Options.file_checksum_gen_factory: %s",
-                   file_checksum_gen_factory
-                       ? file_checksum_gen_factory->Name()
-                       : kUnknownFileChecksumFuncName.c_str());
+                   file_checksum_gen_factory ? file_checksum_gen_factory->Name()
+                                             : kUnknownFileChecksumFuncName);
  ROCKS_LOG_HEADER(log, "                Options.best_efforts_recovery: %d",
                   static_cast<int>(best_efforts_recovery));
 }
@@ -8,6 +8,9 @@ LIB_SOURCES =                                                   \
  db/blob/blob_file_addition.cc                                 \
  db/blob/blob_file_garbage.cc                                  \
  db/blob/blob_file_meta.cc                                     \
+  db/blob/blob_log_format.cc                                    \
+  db/blob/blob_log_reader.cc                                    \
+  db/blob/blob_log_writer.cc                                    \
  db/builder.cc                                                 \
  db/c.cc                                                       \
  db/column_family.cc                                           \
@@ -198,9 +201,6 @@ LIB_SOURCES =                                                   \
  utilities/blob_db/blob_db_impl.cc                             \
  utilities/blob_db/blob_db_impl_filesnapshot.cc                \
  utilities/blob_db/blob_file.cc                                \
-  utilities/blob_db/blob_log_format.cc                          \
-  utilities/blob_db/blob_log_reader.cc                          \
-  utilities/blob_db/blob_log_writer.cc                          \
  utilities/cassandra/cassandra_compaction_filter.cc            \
  utilities/cassandra/format.cc                                 \
  utilities/cassandra/merge_operator.cc                         \
@@ -1781,7 +1781,7 @@ const char* BlockBasedTableBuilder::GetFileChecksumFuncName() const {
  if (rep_->file != nullptr) {
    return rep_->file->GetFileChecksumFuncName();
  } else {
-    return kUnknownFileChecksumFuncName.c_str();
+    return kUnknownFileChecksumFuncName;
  }
 }

@@ -420,7 +420,8 @@ Status BlockBasedTableFactory::NewTableReader(
      table_reader_options.level, table_reader_options.immortal,
      table_reader_options.largest_seqno,
      table_reader_options.force_direct_prefetch, &tail_prefetch_stats_,
-      table_reader_options.block_cache_tracer);
+      table_reader_options.block_cache_tracer,
+      table_reader_options.max_file_size_for_l0_meta_pin);
 }

 TableBuilder* BlockBasedTableFactory::NewTableBuilder(
@@ -610,7 +610,8 @@ Status BlockBasedTable::Open(
    const int level, const bool immortal_table,
    const SequenceNumber largest_seqno, const bool force_direct_prefetch,
    TailPrefetchStats* tail_prefetch_stats,
-    BlockCacheTracer* const block_cache_tracer) {
+    BlockCacheTracer* const block_cache_tracer,
+    size_t max_file_size_for_l0_meta_pin) {
  table_reader->reset();

  Status s;
@@ -706,7 +707,8 @@ Status BlockBasedTable::Open(
  }
  s = new_table->PrefetchIndexAndFilterBlocks(
      prefetch_buffer.get(), metaindex_iter.get(), new_table.get(),
-      prefetch_all, table_options, level, &lookup_context);
+      prefetch_all, table_options, level, file_size,
+      max_file_size_for_l0_meta_pin, &lookup_context);

  if (s.ok()) {
    // Update tail prefetch stats
@@ -943,6 +945,7 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks(
    FilePrefetchBuffer* prefetch_buffer, InternalIterator* meta_iter,
    BlockBasedTable* new_table, bool prefetch_all,
    const BlockBasedTableOptions& table_options, const int level,
+    size_t file_size, size_t max_file_size_for_l0_meta_pin,
    BlockCacheLookupContext* lookup_context) {
  Status s;

@@ -987,9 +990,10 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks(

  const bool use_cache = table_options.cache_index_and_filter_blocks;

-  // pin both index and filters, down to all partitions
+  // pin both index and filters, down to all partitions.
  const bool pin_all =
-      rep_->table_options.pin_l0_filter_and_index_blocks_in_cache && level == 0;
+      rep_->table_options.pin_l0_filter_and_index_blocks_in_cache &&
+      level == 0 && file_size <= max_file_size_for_l0_meta_pin;

  // prefetch the first level of index
  const bool prefetch_index =
@@ -2837,6 +2841,12 @@ Status BlockBasedTable::VerifyChecksumInBlocks(
      break;
    }
  }
+  if (s.ok()) {
+    // In the case of two level indexes, we would have exited the above loop
+    // by checking index_iter->Valid(), but Valid() might have returned false
+    // due to an IO error. So check the index_iter status
+    s = index_iter->status();
+  }
  return s;
 }

@@ -101,7 +101,8 @@ class BlockBasedTable : public TableReader {
                     const SequenceNumber largest_seqno = 0,
                     bool force_direct_prefetch = false,
                     TailPrefetchStats* tail_prefetch_stats = nullptr,
-                     BlockCacheTracer* const block_cache_tracer = nullptr);
+                     BlockCacheTracer* const block_cache_tracer = nullptr,
+                     size_t max_file_size_for_l0_meta_pin = 0);

  bool PrefixMayMatch(const Slice& internal_key,
                      const ReadOptions& read_options,
@@ -255,6 +256,7 @@ class BlockBasedTable : public TableReader {

 private:
  friend class MockedBlockBasedTable;
+  friend class BlockBasedTableReaderTestVerifyChecksum_ChecksumMismatch_Test;
  static std::atomic<uint64_t> next_cache_key_id_;
  BlockCacheTracer* const block_cache_tracer_;

@@ -420,6 +422,7 @@ class BlockBasedTable : public TableReader {
      FilePrefetchBuffer* prefetch_buffer, InternalIterator* meta_iter,
      BlockBasedTable* new_table, bool prefetch_all,
      const BlockBasedTableOptions& table_options, const int level,
+      size_t file_size, size_t max_file_size_for_l0_meta_pin,
      BlockCacheLookupContext* lookup_context);

  static BlockType GetBlockTypeForMetaBlockByName(const Slice& meta_block_name);
@@ -4,6 +4,8 @@
 //  (found in the LICENSE.Apache file in the root directory).

 #include "table/block_based/block_based_table_reader.h"
+#include "rocksdb/file_system.h"
+#include "table/block_based/partitioned_index_iterator.h"

 #include "db/table_properties_collector.h"
 #include "options/options_helper.h"
@@ -19,19 +21,29 @@ namespace ROCKSDB_NAMESPACE {

 class BlockBasedTableReaderTest
    : public testing::Test,
-      public testing::WithParamInterface<std::tuple<CompressionType, bool>> {
+      public testing::WithParamInterface<std::tuple<
+          CompressionType, bool, BlockBasedTableOptions::IndexType, bool>> {
 protected:
  CompressionType compression_type_;
  bool use_direct_reads_;

  void SetUp() override {
-    std::tie(compression_type_, use_direct_reads_) = GetParam();
+    BlockBasedTableOptions::IndexType index_type;
+    bool no_block_cache;
+    std::tie(compression_type_, use_direct_reads_, index_type, no_block_cache) =
+        GetParam();

    test::SetupSyncPointsToMockDirectIO();
    test_dir_ = test::PerThreadDBPath("block_based_table_reader_test");
    env_ = Env::Default();
    fs_ = FileSystem::Default();
    ASSERT_OK(fs_->CreateDir(test_dir_, IOOptions(), nullptr));
+
+    BlockBasedTableOptions opts;
+    opts.index_type = index_type;
+    opts.no_block_cache = no_block_cache;
+    table_factory_.reset(
+        static_cast<BlockBasedTableFactory*>(NewBlockBasedTableFactory(opts)));
  }

  void TearDown() override { EXPECT_OK(test::DestroyDir(env_, test_dir_)); }
@@ -50,7 +62,7 @@ class BlockBasedTableReaderTest
    ColumnFamilyOptions cf_options;
    MutableCFOptions moptions(cf_options);
    std::vector<std::unique_ptr<IntTblPropCollectorFactory>> factories;
-    std::unique_ptr<TableBuilder> table_builder(table_factory_.NewTableBuilder(
+    std::unique_ptr<TableBuilder> table_builder(table_factory_->NewTableBuilder(
        TableBuilderOptions(ioptions, moptions, comparator, &factories,
                            compression_type, 0 /* sample_for_compression */,
                            CompressionOptions(), false /* skip_filters */,
@@ -79,19 +91,21 @@ class BlockBasedTableReaderTest

    std::unique_ptr<TableReader> table_reader;
    ASSERT_OK(BlockBasedTable::Open(ioptions, EnvOptions(),
-                                    table_factory_.table_options(), comparator,
+                                    table_factory_->table_options(), comparator,
                                    std::move(file), file_size, &table_reader));

    table->reset(reinterpret_cast<BlockBasedTable*>(table_reader.release()));
  }

+  std::string Path(const std::string& fname) { return test_dir_ + "/" + fname; }
+
+  const std::shared_ptr<FileSystem>& fs() const { return fs_; }
+
 private:
  std::string test_dir_;
  Env* env_;
  std::shared_ptr<FileSystem> fs_;
-  BlockBasedTableFactory table_factory_;
-
-  std::string Path(const std::string& fname) { return test_dir_ + "/" + fname; }
+  std::unique_ptr<BlockBasedTableFactory> table_factory_;

  void WriteToFile(const std::string& content, const std::string& filename) {
    std::unique_ptr<FSWritableFile> f;
@@ -219,20 +233,104 @@ TEST_P(BlockBasedTableReaderTest, MultiGet) {
  }
 }

+class BlockBasedTableReaderTestVerifyChecksum
+    : public BlockBasedTableReaderTest {
+ public:
+  BlockBasedTableReaderTestVerifyChecksum() : BlockBasedTableReaderTest() {}
+};
+
+TEST_P(BlockBasedTableReaderTestVerifyChecksum, ChecksumMismatch) {
+  // Prepare key-value pairs to occupy multiple blocks.
+  // Each value is 256B, every 16 pairs constitute 1 block.
+  // Adjacent blocks contain values with different compression complexity:
+  // human readable strings are easier to compress than random strings.
+  Random rnd(101);
+  std::map<std::string, std::string> kv;
+  {
+    uint32_t key = 0;
+    for (int block = 0; block < 800; block++) {
+      for (int i = 0; i < 16; i++) {
+        char k[9] = {0};
+        // Internal key is constructed directly from this key,
+        // and internal key size is required to be >= 8 bytes,
+        // so use %08u as the format string.
+        sprintf(k, "%08u", key);
+        std::string v;
+        test::RandomString(&rnd, 256, &v);
+        kv[std::string(k)] = v;
+        key++;
+      }
+    }
+  }
+
+  std::string table_name =
+      "BlockBasedTableReaderTest" + CompressionTypeToString(compression_type_);
+  CreateTable(table_name, compression_type_, kv);
+
+  std::unique_ptr<BlockBasedTable> table;
+  Options options;
+  ImmutableCFOptions ioptions(options);
+  FileOptions foptions;
+  foptions.use_direct_reads = use_direct_reads_;
+  InternalKeyComparator comparator(options.comparator);
+  NewBlockBasedTableReader(foptions, ioptions, comparator, table_name, &table);
+
+  // Use the top level iterator to find the offset/size of the first
+  // 2nd level index block and corrupt the block
+  IndexBlockIter iiter_on_stack;
+  BlockCacheLookupContext context{TableReaderCaller::kUserVerifyChecksum};
+  InternalIteratorBase<IndexValue>* iiter = table->NewIndexIterator(
+      ReadOptions(), /*disable_prefix_seek=*/false, &iiter_on_stack,
+      /*get_context=*/nullptr, &context);
+  std::unique_ptr<InternalIteratorBase<IndexValue>> iiter_unique_ptr;
+  if (iiter != &iiter_on_stack) {
+    iiter_unique_ptr = std::unique_ptr<InternalIteratorBase<IndexValue>>(iiter);
+  }
+  ASSERT_OK(iiter->status());
+  iiter->SeekToFirst();
+  BlockHandle handle = static_cast<ParititionedIndexIterator*>(iiter)
+                           ->index_iter_->value()
+                           .handle;
+  table.reset();
+
+  // Corrupt the block pointed to by handle
+  test::CorruptFile(Path(table_name), static_cast<int>(handle.offset()), 128);
+
+  NewBlockBasedTableReader(foptions, ioptions, comparator, table_name, &table);
+  Status s = table->VerifyChecksum(ReadOptions(),
+                                   TableReaderCaller::kUserVerifyChecksum);
+  ASSERT_EQ(s.code(), Status::kCorruption);
+}
+
 // Param 1: compression type
 // Param 2: whether to use direct reads
+// Param 3: Block Based Table Index type
+// Param 4: BBTO no_block_cache option
 #ifdef ROCKSDB_LITE
 // Skip direct I/O tests in lite mode since direct I/O is unsupported.
 INSTANTIATE_TEST_CASE_P(
    MultiGet, BlockBasedTableReaderTest,
-    ::testing::Combine(::testing::ValuesIn(GetSupportedCompressions()),
-                       ::testing::Values(false)));
+    ::testing::Combine(
+        ::testing::ValuesIn(GetSupportedCompressions()),
+        ::testing::Values(false),
+        ::testing::Values(BlockBasedTableOptions::IndexType::kBinarySearch),
+        ::testing::Values(false)));
 #else   // ROCKSDB_LITE
 INSTANTIATE_TEST_CASE_P(
    MultiGet, BlockBasedTableReaderTest,
-    ::testing::Combine(::testing::ValuesIn(GetSupportedCompressions()),
-                       ::testing::Bool()));
+    ::testing::Combine(
+        ::testing::ValuesIn(GetSupportedCompressions()), ::testing::Bool(),
+        ::testing::Values(BlockBasedTableOptions::IndexType::kBinarySearch),
+        ::testing::Values(false)));
 #endif  // ROCKSDB_LITE
+INSTANTIATE_TEST_CASE_P(
+    VerifyChecksum, BlockBasedTableReaderTestVerifyChecksum,
+    ::testing::Combine(
+        ::testing::ValuesIn(GetSupportedCompressions()),
+        ::testing::Values(false),
+        ::testing::Values(
+            BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch),
+        ::testing::Values(true)));

 }  // namespace ROCKSDB_NAMESPACE

@@ -122,6 +122,7 @@ class ParititionedIndexIterator : public InternalIteratorBase<IndexValue> {
  }

 private:
+  friend class BlockBasedTableReaderTestVerifyChecksum_ChecksumMismatch_Test;
  const BlockBasedTable* table_;
  const ReadOptions read_options_;
 #ifndef NDEBUG
@@ -528,7 +528,7 @@ const char* CuckooTableBuilder::GetFileChecksumFuncName() const {
  if (file_ != nullptr) {
    return file_->GetFileChecksumFuncName();
  } else {
-    return kUnknownFileChecksumFuncName.c_str();
+    return kUnknownFileChecksumFuncName;
  }
 }

@@ -163,7 +163,7 @@ class MockTableBuilder : public TableBuilder {
  std::string GetFileChecksum() const override { return kUnknownFileChecksum; }
  // Get file checksum function name
  const char* GetFileChecksumFuncName() const override {
-    return kUnknownFileChecksumFuncName.c_str();
+    return kUnknownFileChecksumFuncName;
  }

 private:
@@ -313,7 +313,7 @@ const char* PlainTableBuilder::GetFileChecksumFuncName() const {
  if (file_ != nullptr) {
    return file_->GetFileChecksumFuncName();
  } else {
-    return kUnknownFileChecksumFuncName.c_str();
+    return kUnknownFileChecksumFuncName;
  }
 }

@@ -34,11 +34,13 @@ struct TableReaderOptions {
                     const InternalKeyComparator& _internal_comparator,
                     bool _skip_filters = false, bool _immortal = false,
                     bool _force_direct_prefetch = false, int _level = -1,
-                     BlockCacheTracer* const _block_cache_tracer = nullptr)
+                     BlockCacheTracer* const _block_cache_tracer = nullptr,
+                     size_t _max_file_size_for_l0_meta_pin = 0)
      : TableReaderOptions(_ioptions, _prefix_extractor, _env_options,
                           _internal_comparator, _skip_filters, _immortal,
                           _force_direct_prefetch, _level,
-                           0 /* _largest_seqno */, _block_cache_tracer) {}
+                           0 /* _largest_seqno */, _block_cache_tracer,
+                           _max_file_size_for_l0_meta_pin) {}

  // @param skip_filters Disables loading/accessing the filter block
  TableReaderOptions(const ImmutableCFOptions& _ioptions,
@@ -48,7 +50,8 @@ struct TableReaderOptions {
                     bool _skip_filters, bool _immortal,
                     bool _force_direct_prefetch, int _level,
                     SequenceNumber _largest_seqno,
-                     BlockCacheTracer* const _block_cache_tracer)
+                     BlockCacheTracer* const _block_cache_tracer,
+                     size_t _max_file_size_for_l0_meta_pin)
      : ioptions(_ioptions),
        prefix_extractor(_prefix_extractor),
        env_options(_env_options),
@@ -58,7 +61,8 @@ struct TableReaderOptions {
        force_direct_prefetch(_force_direct_prefetch),
        level(_level),
        largest_seqno(_largest_seqno),
-        block_cache_tracer(_block_cache_tracer) {}
+        block_cache_tracer(_block_cache_tracer),
+        max_file_size_for_l0_meta_pin(_max_file_size_for_l0_meta_pin) {}

  const ImmutableCFOptions& ioptions;
  const SliceTransform* prefix_extractor;
@@ -77,6 +81,9 @@ struct TableReaderOptions {
  // largest seqno in the table
  SequenceNumber largest_seqno;
  BlockCacheTracer* const block_cache_tracer;
+  // Largest L0 file size whose meta-blocks may be pinned (can be zero when
+  // unknown).
+  const size_t max_file_size_for_l0_meta_pin;
 };

 struct TableBuilderOptions {
@@ -378,7 +378,8 @@ class TableConstructor: public Constructor {
    return ioptions.table_factory->NewTableReader(
        TableReaderOptions(ioptions, moptions.prefix_extractor.get(), soptions,
                           internal_comparator, !kSkipFilters, !kImmortal,
-                           false, level_, largest_seqno_, &block_cache_tracer_),
+                           false, level_, largest_seqno_, &block_cache_tracer_,
+                           moptions.write_buffer_size),
        std::move(file_reader_), TEST_GetSink()->contents().size(),
        &table_reader_);
  }
@@ -3265,17 +3266,9 @@ TEST_P(BlockBasedTableTest, NoFileChecksum) {
  BlockBasedTableOptions table_options = GetBlockBasedTableOptions();
  std::unique_ptr<InternalKeyComparator> comparator(
      new InternalKeyComparator(BytewiseComparator()));
-  SequenceNumber largest_seqno = 0;
  int level = 0;
  std::vector<std::unique_ptr<IntTblPropCollectorFactory>>
      int_tbl_prop_collector_factories;
-
-  if (largest_seqno != 0) {
-    // Pretend that it's an external file written by SstFileWriter.
-    int_tbl_prop_collector_factories.emplace_back(
-        new SstFileWriterPropertiesCollectorFactory(2 /* version */,
-                                                    0 /* global_seqno*/));
-  }
  std::string column_family_name;

  FileChecksumTestHelper f(true);
@@ -3292,9 +3285,8 @@ TEST_P(BlockBasedTableTest, NoFileChecksum) {
  f.ResetTableBuilder(std::move(builder));
  f.AddKVtoKVMap(1000);
  f.WriteKVAndFlushTable();
-  ASSERT_STREQ(f.GetFileChecksumFuncName(),
-               kUnknownFileChecksumFuncName.c_str());
-  ASSERT_STREQ(f.GetFileChecksum().c_str(), kUnknownFileChecksum.c_str());
+  ASSERT_STREQ(f.GetFileChecksumFuncName(), kUnknownFileChecksumFuncName);
+  ASSERT_STREQ(f.GetFileChecksum().c_str(), kUnknownFileChecksum);
 }

 TEST_P(BlockBasedTableTest, Crc32cFileChecksum) {
@@ -3307,17 +3299,9 @@ TEST_P(BlockBasedTableTest, Crc32cFileChecksum) {
  BlockBasedTableOptions table_options = GetBlockBasedTableOptions();
  std::unique_ptr<InternalKeyComparator> comparator(
      new InternalKeyComparator(BytewiseComparator()));
-  SequenceNumber largest_seqno = 0;
  int level = 0;
  std::vector<std::unique_ptr<IntTblPropCollectorFactory>>
      int_tbl_prop_collector_factories;
-
-  if (largest_seqno != 0) {
-    // Pretend that it's an external file written by SstFileWriter.
-    int_tbl_prop_collector_factories.emplace_back(
-        new SstFileWriterPropertiesCollectorFactory(2 /* version */,
-                                                    0 /* global_seqno*/));
-  }
  std::string column_family_name;

  FileChecksumGenContext gen_context;
@@ -3446,9 +3430,8 @@ TEST_F(PlainTableTest, NoFileChecksum) {
  f.ResetTableBuilder(std::move(builder));
  f.AddKVtoKVMap(1000);
  f.WriteKVAndFlushTable();
-  ASSERT_STREQ(f.GetFileChecksumFuncName(),
-               kUnknownFileChecksumFuncName.c_str());
-  EXPECT_EQ(f.GetFileChecksum(), kUnknownFileChecksum.c_str());
+  ASSERT_STREQ(f.GetFileChecksumFuncName(), kUnknownFileChecksumFuncName);
+  EXPECT_EQ(f.GetFileChecksum(), kUnknownFileChecksum);
 }

 TEST_F(PlainTableTest, Crc32cFileChecksum) {
@@ -10,6 +10,7 @@
 #include "test_util/testutil.h"

 #include <fcntl.h>
+#include <sys/stat.h>
 #include <array>
 #include <cctype>
 #include <fstream>
@@ -464,7 +465,16 @@ Status DestroyDir(Env* env, const std::string& dir) {
      if (file_in_dir == "." || file_in_dir == "..") {
        continue;
      }
-      s = env->DeleteFile(dir + "/" + file_in_dir);
+      std::string path = dir + "/" + file_in_dir;
+      bool is_dir = false;
+      s = env->IsDirectory(path, &is_dir);
+      if (s.ok()) {
+        if (is_dir) {
+          s = DestroyDir(env, path);
+        } else {
+          s = env->DeleteFile(path);
+        }
+      }
      if (!s.ok()) {
        break;
      }
@@ -527,5 +537,46 @@ void SetupSyncPointsToMockDirectIO() {
 #endif
 }

+void CorruptFile(const std::string& fname, int offset, int bytes_to_corrupt) {
+  struct stat sbuf;
+  if (stat(fname.c_str(), &sbuf) != 0) {
+    // strerror is not thread-safe so should not be used in the "passing" path
+    // of unit tests (sometimes parallelized) but is OK here where test fails
+    const char* msg = strerror(errno);
+    fprintf(stderr, "%s:%s\n", fname.c_str(), msg);
+    assert(false);
+  }
+
+  if (offset < 0) {
+    // Relative to end of file; make it absolute
+    if (-offset > sbuf.st_size) {
+      offset = 0;
+    } else {
+      offset = static_cast<int>(sbuf.st_size + offset);
+    }
+  }
+  if (offset > sbuf.st_size) {
+    offset = static_cast<int>(sbuf.st_size);
+  }
+  if (offset + bytes_to_corrupt > sbuf.st_size) {
+    bytes_to_corrupt = static_cast<int>(sbuf.st_size - offset);
+  }
+
+  // Do it
+  std::string contents;
+  Status s = ReadFileToString(Env::Default(), fname, &contents);
+  assert(s.ok());
+  for (int i = 0; i < bytes_to_corrupt; i++) {
+    contents[i + offset] ^= 0x80;
+  }
+  s = WriteStringToFile(Env::Default(), contents, fname);
+  assert(s.ok());
+  Options options;
+  EnvOptions env_options;
+#ifndef ROCKSDB_LITE
+  assert(!VerifySstFileChecksum(options, env_options, fname).ok());
+#endif
+}
+
 }  // namespace test
 }  // namespace ROCKSDB_NAMESPACE
@@ -812,5 +812,7 @@ void ResetTmpDirForDirectIO();
 // to the file system.
 void SetupSyncPointsToMockDirectIO();

+void CorruptFile(const std::string& fname, int offset, int bytes_to_corrupt);
+
 }  // namespace test
 }  // namespace ROCKSDB_NAMESPACE
@@ -0,0 +1,42 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "test_util/testutil.h"
+
+#include "port/port.h"
+#include "port/stack_trace.h"
+#include "test_util/testharness.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+void CreateFile(Env* env, const std::string& path) {
+  std::unique_ptr<WritableFile> f;
+  ASSERT_OK(env->NewWritableFile(path, &f, EnvOptions()));
+  f->Close();
+}
+
+TEST(TestUtil, DestroyDirRecursively) {
+  auto env = Env::Default();
+  // test_util/file
+  //          /dir
+  //          /dir/file
+  std::string test_dir = test::PerThreadDBPath("test_util");
+  ASSERT_OK(env->CreateDir(test_dir));
+  CreateFile(env, test_dir + "/file");
+  ASSERT_OK(env->CreateDir(test_dir + "/dir"));
+  CreateFile(env, test_dir + "/dir/file");
+
+  ASSERT_OK(test::DestroyDir(env, test_dir));
+  auto s = env->FileExists(test_dir);
+  ASSERT_TRUE(s.IsNotFound());
+}
+
+}  // namespace ROCKSDB_NAMESPACE
+
+int main(int argc, char** argv) {
+  ROCKSDB_NAMESPACE::port::InstallStackTraceHandler();
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
@@ -97,7 +97,7 @@ void DumpSstFile(Options options, std::string filename, bool output_hex,
 };

 LDBCommand* LDBCommand::InitFromCmdLineArgs(
-    int argc, char** argv, const Options& options,
+    int argc, char const* const* argv, const Options& options,
    const LDBOptions& ldb_options,
    const std::vector<ColumnFamilyDescriptor>* column_families) {
  std::vector<std::string> args;
@@ -41,6 +41,18 @@ class LdbCmdTest : public testing::Test {
  std::shared_ptr<Env> env_guard_;
 };

+TEST_F(LdbCmdTest, HelpAndVersion) {
+  Options o;
+  o.env = TryLoadCustomOrDefaultEnv();
+  LDBOptions lo;
+  static const char* help[] = {"./ldb", "--help"};
+  ASSERT_EQ(0, LDBCommandRunner::RunCommand(2, help, o, lo, nullptr));
+  static const char* version[] = {"./ldb", "--version"};
+  ASSERT_EQ(0, LDBCommandRunner::RunCommand(2, version, o, lo, nullptr));
+  static const char* bad[] = {"./ldb", "--not_an_option"};
+  ASSERT_NE(0, LDBCommandRunner::RunCommand(2, bad, o, lo, nullptr));
+}
+
 TEST_F(LdbCmdTest, HexToString) {
  // map input to expected outputs.
  // odd number of "hex" half bytes doesn't make sense
@@ -13,7 +13,7 @@ namespace ROCKSDB_NAMESPACE {
 LDBOptions::LDBOptions() {}

 void LDBCommandRunner::PrintHelp(const LDBOptions& ldb_options,
-                                 const char* /*exec_name*/) {
+                                 const char* /*exec_name*/, bool to_stderr) {
  std::string ret;

  ret.append(ldb_options.print_help_header);
@@ -99,22 +99,32 @@ void LDBCommandRunner::PrintHelp(const LDBOptions& ldb_options,
  WriteExternalSstFilesCommand::Help(ret);
  IngestExternalSstFilesCommand::Help(ret);

-  fprintf(stderr, "%s\n", ret.c_str());
+  fprintf(to_stderr ? stderr : stdout, "%s\n", ret.c_str());
 }

 int LDBCommandRunner::RunCommand(
-    int argc, char** argv, Options options, const LDBOptions& ldb_options,
+    int argc, char const* const* argv, Options options,
+    const LDBOptions& ldb_options,
    const std::vector<ColumnFamilyDescriptor>* column_families) {
  if (argc <= 2) {
-    PrintHelp(ldb_options, argv[0]);
-    return 1;
+    if (std::string(argv[1]) == "--version") {
+      printf("ldb from RocksDB %d.%d.%d\n", ROCKSDB_MAJOR, ROCKSDB_MINOR,
+             ROCKSDB_PATCH);
+      return 0;
+    } else if (std::string(argv[1]) == "--help") {
+      PrintHelp(ldb_options, argv[0], /*to_stderr*/ false);
+      return 0;
+    } else {
+      PrintHelp(ldb_options, argv[0], /*to_stderr*/ true);
+      return 1;
+    }
  }

  LDBCommand* cmdObj = LDBCommand::InitFromCmdLineArgs(
      argc, argv, options, ldb_options, column_families);
  if (cmdObj == nullptr) {
    fprintf(stderr, "Unknown command\n");
-    PrintHelp(ldb_options, argv[0]);
+    PrintHelp(ldb_options, argv[0], /*to_stderr*/ true);
    return 1;
  }

@@ -9,8 +9,7 @@

 int main(int argc, char** argv) {
  ROCKSDB_NAMESPACE::SSTDumpTool tool;
-  tool.Run(argc, argv);
-  return 0;
+  return tool.Run(argc, argv);
 }
 #else
 #include <stdio.h>
@@ -129,6 +129,20 @@ class SSTDumpToolTest : public testing::Test {
  }
 };

+TEST_F(SSTDumpToolTest, HelpAndVersion) {
+  Options opts;
+  opts.env = env();
+
+  ROCKSDB_NAMESPACE::SSTDumpTool tool;
+
+  static const char* help[] = {"./sst_dump", "--help"};
+  ASSERT_TRUE(!tool.Run(2, help, opts));
+  static const char* version[] = {"./sst_dump", "--version"};
+  ASSERT_TRUE(!tool.Run(2, version, opts));
+  static const char* bad[] = {"./sst_dump", "--not_an_option"};
+  ASSERT_TRUE(tool.Run(2, bad, opts));
+}
+
 TEST_F(SSTDumpToolTest, EmptyFilter) {
  Options opts;
  opts.env = env();
@@ -298,13 +312,50 @@ TEST_F(SSTDumpToolTest, NoSstFile) {
        "--command=verify", "--command=recompress", "--command=verify_checksum",
        "--show_properties"}) {
    snprintf(usage[1], kOptLength, "%s", command);
-    ASSERT_TRUE(!tool.Run(3, usage, opts));
+    ASSERT_TRUE(tool.Run(3, usage, opts));
  }
  for (int i = 0; i < 3; i++) {
    delete[] usage[i];
  }
 }

+TEST_F(SSTDumpToolTest, ValidSSTPath) {
+  Options opts;
+  opts.env = env();
+  char* usage[3];
+  PopulateCommandArgs("", "", usage);
+  SSTDumpTool tool;
+  std::string file_not_exists = MakeFilePath("file_not_exists.sst");
+  std::string sst_file = MakeFilePath("rocksdb_sst_test.sst");
+  createSST(opts, sst_file);
+  std::string text_file = MakeFilePath("text_file");
+  ASSERT_OK(WriteStringToFile(opts.env, "Hello World!", text_file));
+  std::string fake_sst = MakeFilePath("fake_sst.sst");
+  ASSERT_OK(WriteStringToFile(opts.env, "Not an SST file!", fake_sst));
+
+  for (const auto& command_arg : {"--command=verify", "--command=identify"}) {
+    snprintf(usage[1], kOptLength, "%s", command_arg);
+
+    snprintf(usage[2], kOptLength, "--file=%s", file_not_exists.c_str());
+    ASSERT_TRUE(tool.Run(3, usage, opts));
+
+    snprintf(usage[2], kOptLength, "--file=%s", sst_file.c_str());
+    ASSERT_TRUE(!tool.Run(3, usage, opts));
+
+    snprintf(usage[2], kOptLength, "--file=%s", text_file.c_str());
+    ASSERT_TRUE(tool.Run(3, usage, opts));
+
+    snprintf(usage[2], kOptLength, "--file=%s", fake_sst.c_str());
+    ASSERT_TRUE(tool.Run(3, usage, opts));
+  }
+  ASSERT_OK(opts.env->DeleteFile(sst_file));
+  ASSERT_OK(opts.env->DeleteFile(text_file));
+  ASSERT_OK(opts.env->DeleteFile(fake_sst));
+
+  for (int i = 0; i < 3; i++) {
+    delete[] usage[i];
+  }
+}
 }  // namespace ROCKSDB_NAMESPACE

 #ifdef ROCKSDB_UNITTESTS_WITH_CUSTOM_OBJECTS_FROM_STATIC_LIBS
@@ -95,6 +95,12 @@ Status SstFileDumper::GetTableReader(const std::string& file_path) {
    s = options_.env->GetFileSize(file_path, &file_size);
  }

+  // check empty file
+  // if true, skip further processing of this file
+  if (file_size == 0) {
+    return Status::Aborted(file_path, "Empty file");
+  }
+
  file_.reset(new RandomAccessFileReader(NewLegacyRandomAccessFileWrapper(file),
                                         file_path));

@@ -475,23 +481,24 @@ Status SstFileDumper::ReadTableProperties(

 namespace {

-void print_help() {
+void print_help(bool to_stderr) {
  fprintf(
-      stderr,
-      R"(sst_dump --file=<data_dir_OR_sst_file> [--command=check|scan|raw|recompress]
+      to_stderr ? stderr : stdout,
+      R"(sst_dump --file=<data_dir_OR_sst_file> [--command=check|scan|raw|recompress|identify]
    --file=<data_dir_OR_sst_file>
      Path to SST file or directory containing SST files

    --env_uri=<uri of underlying Env>
      URI of underlying Env

-    --command=check|scan|raw|verify
+    --command=check|scan|raw|verify|identify
        check: Iterate over entries in files but don't print anything except if an error is encountered (default command)
        scan: Iterate over entries in files and print them to screen
        raw: Dump all the table contents to <file_name>_dump.txt
        verify: Iterate all the blocks in files verifying checksum to detect possible corruption but don't print anything except if a corruption is encountered
        recompress: reports the SST file size if recompressed with different
                    compression types
+        identify: Reports a file is a valid SST file or lists all valid SST files under a directory

    --output_hex
      Can be combined with scan command to print the keys and values in Hex
@@ -520,7 +527,7 @@ void print_help() {

    --show_properties
      Print table properties after iterating over the file when executing
-      check|scan|raw
+      check|scan|raw|identify

    --set_block_size=<block_size>
      Can be combined with --command=recompress to set the block size that will
@@ -565,7 +572,7 @@ bool ParseIntArg(const char* arg, const std::string arg_name,
 }
 }  // namespace

-int SSTDumpTool::Run(int argc, char** argv, Options options) {
+int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
  const char* env_uri = nullptr;
  const char* dir_or_file = nullptr;
  uint64_t read_num = std::numeric_limits<uint64_t>::max();
@@ -689,10 +696,17 @@ int SSTDumpTool::Run(int argc, char** argv, Options options) {
                           "compression_level_to must be numeric", &tmp_val)) {
      has_compression_level_to = true;
      compress_level_to = static_cast<int>(tmp_val);
+    } else if (strcmp(argv[i], "--help") == 0) {
+      print_help(/*to_stderr*/ false);
+      return 0;
+    } else if (strcmp(argv[i], "--version") == 0) {
+      printf("sst_dump from RocksDB %d.%d.%d\n", ROCKSDB_MAJOR, ROCKSDB_MINOR,
+             ROCKSDB_PATCH);
+      return 0;
    } else {
      fprintf(stderr, "Unrecognized argument '%s'\n\n", argv[i]);
-      print_help();
-      exit(1);
+      print_help(/*to_stderr*/ true);
+      return 1;
    }
  }

@@ -723,7 +737,7 @@ int SSTDumpTool::Run(int argc, char** argv, Options options) {

  if (dir_or_file == nullptr) {
    fprintf(stderr, "file or directory must be specified.\n\n");
-    print_help();
+    print_help(/*to_stderr*/ true);
    exit(1);
  }

@@ -748,17 +762,29 @@ int SSTDumpTool::Run(int argc, char** argv, Options options) {
  ROCKSDB_NAMESPACE::Env* env = options.env;
  ROCKSDB_NAMESPACE::Status st = env->GetChildren(dir_or_file, &filenames);
  bool dir = true;
-  if (!st.ok()) {
+  if (!st.ok() || filenames.empty()) {
+    // dir_or_file does not exist or does not contain children
+    // Check its existence first
+    Status s = env->FileExists(dir_or_file);
+    // dir_or_file does not exist
+    if (!s.ok()) {
+      fprintf(stderr, "%s%s: No such file or directory\n", s.ToString().c_str(),
+              dir_or_file);
+      return 1;
+    }
+    // dir_or_file exists and is treated as a "file"
+    // since it has no children
+    // This is ok since later it will be checked
+    // that whether it is a valid sst or not
+    // (A directory "file" is not a valid sst)
    filenames.clear();
    filenames.push_back(dir_or_file);
    dir = false;
  }

-  fprintf(stdout, "from [%s] to [%s]\n",
-          ROCKSDB_NAMESPACE::Slice(from_key).ToString(true).c_str(),
-          ROCKSDB_NAMESPACE::Slice(to_key).ToString(true).c_str());
-
  uint64_t total_read = 0;
+  // List of RocksDB SST file without corruption
+  std::vector<std::string> valid_sst_files;
  for (size_t i = 0; i < filenames.size(); i++) {
    std::string filename = filenames.at(i);
    if (filename.length() <= 4 ||
@@ -766,6 +792,7 @@ int SSTDumpTool::Run(int argc, char** argv, Options options) {
      // ignore
      continue;
    }
+
    if (dir) {
      filename = std::string(dir_or_file) + "/" + filename;
    }
@@ -773,10 +800,23 @@ int SSTDumpTool::Run(int argc, char** argv, Options options) {
    ROCKSDB_NAMESPACE::SstFileDumper dumper(options, filename, readahead_size,
                                            verify_checksum, output_hex,
                                            decode_blob_index);
+    // Not a valid SST
    if (!dumper.getStatus().ok()) {
      fprintf(stderr, "%s: %s\n", filename.c_str(),
              dumper.getStatus().ToString().c_str());
      continue;
+    } else {
+      valid_sst_files.push_back(filename);
+      // Print out from and to key information once
+      // where there is at least one valid SST
+      if (valid_sst_files.size() == 1) {
+        // from_key and to_key are only used for "check", "scan", or ""
+        if (command == "check" || command == "scan" || command == "") {
+          fprintf(stdout, "from [%s] to [%s]\n",
+                  ROCKSDB_NAMESPACE::Slice(from_key).ToString(true).c_str(),
+                  ROCKSDB_NAMESPACE::Slice(to_key).ToString(true).c_str());
+        }
+      }
    }

    if (command == "recompress") {
@@ -881,7 +921,35 @@ int SSTDumpTool::Run(int argc, char** argv, Options options) {
    fprintf(stdout, "total filter block size: %" PRIu64 "\n",
            total_filter_block_size);
  }
-  return 0;
+
+  if (valid_sst_files.empty()) {
+    // No valid SST files are found
+    // Exit with an error state
+    if (dir) {
+      fprintf(stdout, "------------------------------\n");
+      fprintf(stderr, "No valid SST files found in %s\n", dir_or_file);
+    } else {
+      fprintf(stderr, "%s is not a valid SST file\n", dir_or_file);
+    }
+    return 1;
+  } else {
+    if (command == "identify") {
+      if (dir) {
+        fprintf(stdout, "------------------------------\n");
+        fprintf(stdout, "List of valid SST files found in %s:\n", dir_or_file);
+        for (const auto& f : valid_sst_files) {
+          fprintf(stdout, "%s\n", f.c_str());
+        }
+        fprintf(stdout, "Number of valid SST files: %zu\n",
+                valid_sst_files.size());
+      } else {
+        fprintf(stdout, "%s is a valid SST file\n", dir_or_file);
+      }
+    }
+    // At least one valid SST
+    // exit with a success state
+    return 0;
+  }
 }
 }  // namespace ROCKSDB_NAMESPACE

@@ -188,7 +188,10 @@ TEST_F(TraceAnalyzerTest, Get) {
  std::string trace_path = test_path_ + "/trace";
  std::string output_path = test_path_ + "/get";
  std::string file_path;
-  std::vector<std::string> paras = {"-analyze_get"};
+  std::vector<std::string> paras = {
+      "-analyze_get=true",           "-analyze_put=false",
+      "-analyze_delete=false",       "-analyze_single_delete=false",
+      "-analyze_range_delete=false", "-analyze_iterator=false"};
  paras.push_back("-output_dir=" + output_path);
  paras.push_back("-trace_path=" + trace_path);
  paras.push_back("-key_space_dir=" + test_path_);
@@ -254,7 +257,10 @@ TEST_F(TraceAnalyzerTest, Put) {
  std::string trace_path = test_path_ + "/trace";
  std::string output_path = test_path_ + "/put";
  std::string file_path;
-  std::vector<std::string> paras = {"-analyze_put"};
+  std::vector<std::string> paras = {
+      "-analyze_get=false",          "-analyze_put=true",
+      "-analyze_delete=false",       "-analyze_single_delete=false",
+      "-analyze_range_delete=false", "-analyze_iterator=false"};
  paras.push_back("-output_dir=" + output_path);
  paras.push_back("-trace_path=" + trace_path);
  paras.push_back("-key_space_dir=" + test_path_);
@@ -298,7 +304,7 @@ TEST_F(TraceAnalyzerTest, Put) {
  CheckFileContent(k_whole_prefix, file_path, true);

  // Check the overall qps
-  std::vector<std::string> all_qps = {"1 1 0 0 0 0 0 0 2"};
+  std::vector<std::string> all_qps = {"0 1 0 0 0 0 0 0 1"};
  file_path = output_path + "/test-qps_stats.txt";
  CheckFileContent(all_qps, file_path, true);

@@ -325,7 +331,10 @@ TEST_F(TraceAnalyzerTest, Delete) {
  std::string trace_path = test_path_ + "/trace";
  std::string output_path = test_path_ + "/delete";
  std::string file_path;
-  std::vector<std::string> paras = {"-analyze_delete"};
+  std::vector<std::string> paras = {
+      "-analyze_get=false",          "-analyze_put=false",
+      "-analyze_delete=true",        "-analyze_single_delete=false",
+      "-analyze_range_delete=false", "-analyze_iterator=false"};
  paras.push_back("-output_dir=" + output_path);
  paras.push_back("-trace_path=" + trace_path);
  paras.push_back("-key_space_dir=" + test_path_);
@@ -370,7 +379,7 @@ TEST_F(TraceAnalyzerTest, Delete) {
  CheckFileContent(k_whole_prefix, file_path, true);

  // Check the overall qps
-  std::vector<std::string> all_qps = {"1 1 1 0 0 0 0 0 3"};
+  std::vector<std::string> all_qps = {"0 0 1 0 0 0 0 0 1"};
  file_path = output_path + "/test-qps_stats.txt";
  CheckFileContent(all_qps, file_path, true);

@@ -391,7 +400,11 @@ TEST_F(TraceAnalyzerTest, Merge) {
  std::string trace_path = test_path_ + "/trace";
  std::string output_path = test_path_ + "/merge";
  std::string file_path;
-  std::vector<std::string> paras = {"-analyze_merge"};
+  std::vector<std::string> paras = {
+      "-analyze_get=false",           "-analyze_put=false",
+      "-analyze_delete=false",        "-analyze_merge=true",
+      "-analyze_single_delete=false", "-analyze_range_delete=false",
+      "-analyze_iterator=false"};
  paras.push_back("-output_dir=" + output_path);
  paras.push_back("-trace_path=" + trace_path);
  paras.push_back("-key_space_dir=" + test_path_);
@@ -435,7 +448,7 @@ TEST_F(TraceAnalyzerTest, Merge) {
  CheckFileContent(k_whole_prefix, file_path, true);

  // Check the overall qps
-  std::vector<std::string> all_qps = {"1 1 1 0 0 1 0 0 4"};
+  std::vector<std::string> all_qps = {"0 0 0 0 0 1 0 0 1"};
  file_path = output_path + "/test-qps_stats.txt";
  CheckFileContent(all_qps, file_path, true);

@@ -463,7 +476,11 @@ TEST_F(TraceAnalyzerTest, SingleDelete) {
  std::string trace_path = test_path_ + "/trace";
  std::string output_path = test_path_ + "/single_delete";
  std::string file_path;
-  std::vector<std::string> paras = {"-analyze_single_delete"};
+  std::vector<std::string> paras = {
+      "-analyze_get=false",          "-analyze_put=false",
+      "-analyze_delete=false",       "-analyze_merge=false",
+      "-analyze_single_delete=true", "-analyze_range_delete=false",
+      "-analyze_iterator=false"};
  paras.push_back("-output_dir=" + output_path);
  paras.push_back("-trace_path=" + trace_path);
  paras.push_back("-key_space_dir=" + test_path_);
@@ -508,7 +525,7 @@ TEST_F(TraceAnalyzerTest, SingleDelete) {
  CheckFileContent(k_whole_prefix, file_path, true);

  // Check the overall qps
-  std::vector<std::string> all_qps = {"1 1 1 1 0 1 0 0 5"};
+  std::vector<std::string> all_qps = {"0 0 0 1 0 0 0 0 1"};
  file_path = output_path + "/test-qps_stats.txt";
  CheckFileContent(all_qps, file_path, true);

@@ -530,7 +547,11 @@ TEST_F(TraceAnalyzerTest, DeleteRange) {
  std::string trace_path = test_path_ + "/trace";
  std::string output_path = test_path_ + "/range_delete";
  std::string file_path;
-  std::vector<std::string> paras = {"-analyze_range_delete"};
+  std::vector<std::string> paras = {
+      "-analyze_get=false",           "-analyze_put=false",
+      "-analyze_delete=false",        "-analyze_merge=false",
+      "-analyze_single_delete=false", "-analyze_range_delete=true",
+      "-analyze_iterator=false"};
  paras.push_back("-output_dir=" + output_path);
  paras.push_back("-trace_path=" + trace_path);
  paras.push_back("-key_space_dir=" + test_path_);
@@ -576,7 +597,7 @@ TEST_F(TraceAnalyzerTest, DeleteRange) {
  CheckFileContent(k_whole_prefix, file_path, true);

  // Check the overall qps
-  std::vector<std::string> all_qps = {"1 1 1 1 2 1 0 0 7"};
+  std::vector<std::string> all_qps = {"0 0 0 0 2 0 0 0 2"};
  file_path = output_path + "/test-qps_stats.txt";
  CheckFileContent(all_qps, file_path, true);

@@ -599,7 +620,11 @@ TEST_F(TraceAnalyzerTest, Iterator) {
  std::string trace_path = test_path_ + "/trace";
  std::string output_path = test_path_ + "/iterator";
  std::string file_path;
-  std::vector<std::string> paras = {"-analyze_iterator"};
+  std::vector<std::string> paras = {
+      "-analyze_get=false",           "-analyze_put=false",
+      "-analyze_delete=false",        "-analyze_merge=false",
+      "-analyze_single_delete=false", "-analyze_range_delete=false",
+      "-analyze_iterator=true"};
  paras.push_back("-output_dir=" + output_path);
  paras.push_back("-trace_path=" + trace_path);
  paras.push_back("-key_space_dir=" + test_path_);
@@ -645,7 +670,7 @@ TEST_F(TraceAnalyzerTest, Iterator) {
  CheckFileContent(k_whole_prefix, file_path, true);

  // Check the overall qps
-  std::vector<std::string> all_qps = {"1 1 1 1 2 1 1 1 9"};
+  std::vector<std::string> all_qps = {"0 0 0 0 0 0 1 1 2"};
  file_path = output_path + "/test-qps_stats.txt";
  CheckFileContent(all_qps, file_path, true);

@@ -19,6 +19,9 @@
 #include <utility>
 #include <vector>

+#include "db/blob/blob_log_format.h"
+#include "db/blob/blob_log_reader.h"
+#include "db/blob/blob_log_writer.h"
 #include "db/db_iter.h"
 #include "rocksdb/compaction_filter.h"
 #include "rocksdb/db.h"
@@ -30,9 +33,6 @@
 #include "util/timer_queue.h"
 #include "utilities/blob_db/blob_db.h"
 #include "utilities/blob_db/blob_file.h"
-#include "utilities/blob_db/blob_log_format.h"
-#include "utilities/blob_db/blob_log_reader.h"
-#include "utilities/blob_db/blob_log_writer.h"

 namespace ROCKSDB_NAMESPACE {

@@ -8,10 +8,11 @@
 #include <memory>
 #include <string>
 #include <utility>
+
+#include "db/blob/blob_log_format.h"
 #include "file/random_access_file_reader.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/status.h"
-#include "utilities/blob_db/blob_log_format.h"

 namespace ROCKSDB_NAMESPACE {
 namespace blob_db {
@@ -10,13 +10,13 @@
 #include <memory>
 #include <unordered_set>

+#include "db/blob/blob_log_format.h"
+#include "db/blob/blob_log_reader.h"
+#include "db/blob/blob_log_writer.h"
 #include "file/random_access_file_reader.h"
 #include "port/port.h"
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
-#include "utilities/blob_db/blob_log_format.h"
-#include "utilities/blob_db/blob_log_reader.h"
-#include "utilities/blob_db/blob_log_writer.h"

 namespace ROCKSDB_NAMESPACE {
 namespace blob_db {
Author	SHA1	Message	Date
Andrew Kryczka	02db03af8d	make L0 index/filter pinned memory usage predictable (#6911 ) Summary: Memory pinned by `pin_l0_filter_and_index_blocks_in_cache` needs to be predictable based on user config. This PR makes sure we do not pin extra memory for large files generated by intra-L0 (see https://github.com/facebook/rocksdb/issues/6889). Pull Request resolved: https://github.com/facebook/rocksdb/pull/6911 Test Plan: unit test Reviewed By: siying Differential Revision: D21835818 Pulled By: ajkr fbshipit-source-id: a11a088549d06bed8aacc2548d266e5983f0ead4	2020-06-09 16:51:23 -07:00
Levi Tamasi	5abda3bb8b	Move blob_log_{format,reader,writer}.{cc,h} to db/blob/ (#6960 ) Summary: Pull Request resolved: https://github.com/facebook/rocksdb/pull/6960 Test Plan: `make check` Reviewed By: zhichao-cao Differential Revision: D21958416 Pulled By: ltamasi fbshipit-source-id: 97cf05027b7363014b07836e7f158c23827bd661	2020-06-09 15:16:05 -07:00
Peter Dillinger	edf74d1cb1	Add --version and --help to ldb and sst_dump (#6951 ) Summary: as title Pull Request resolved: https://github.com/facebook/rocksdb/pull/6951 Test Plan: tests included + manual Reviewed By: ajkr Differential Revision: D21918540 Pulled By: pdillinger fbshipit-source-id: 79d4991f2a831214fc7e477a839ec19dbbace6c5	2020-06-09 10:04:01 -07:00
sdong	6a8ddd374d	Introduce some Linux build to CircleCI (#6937 ) Summary: Moving towards the long term goal of moving most CI build to CircleCI when possible, add some Linux tests in CircleCI. This is not all what we can include to CircleCI. For example, Java builds are not includ ed. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6937 Test Plan: Watch CI build results. Reviewed By: pdillinger Differential Revision: D21941605 fbshipit-source-id: db6aead3c45f523386d4fb30d224cfde573cccad	2020-06-08 19:34:31 -07:00
anand76	1fb3593f25	Fix a bug in looking up duplicate keys with MultiGet (#6953 ) Summary: When MultiGet is called with duplicate keys, and the key matches the largest key in an SST file and the value type is merge, only the first instance of the duplicate key is returned with correct results. This is due to the incorrect assumption that if a key in a batch is equal to the largest key in the file, the next key cannot be present in that file. Tests: Add a new unit test Pull Request resolved: https://github.com/facebook/rocksdb/pull/6953 Reviewed By: cheng-chang Differential Revision: D21935898 Pulled By: anand1976 fbshipit-source-id: a2cc327a15150e23fd997546ca64d1c33021cb4c	2020-06-08 16:11:21 -07:00
Levi Tamasi	f5e649453c	Add convenience method GetFileMetaDataByNumber (#6940 ) Summary: The patch adds a convenience method `GetFileMetaDataByNumber` that builds on the `FileLocation` functionality introduced recently (see https://github.com/facebook/rocksdb/pull/6862). This method makes it possible to retrieve the `FileMetaData` directly as opposed to having to go through `LevelFiles` and friends. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6940 Test Plan: `make check` Reviewed By: cheng-chang Differential Revision: D21905946 Pulled By: ltamasi fbshipit-source-id: af99e19de21242b2b4a87594a535c6028d16ee72	2020-06-08 16:01:59 -07:00
Zitan Chen	119b26fac0	Implement a new subcommand "identify" for sst_dump (#6943 ) Summary: Implemented a subcommand of sst_dump called identify, which determines whether a file is an SST file or identifies and lists all the SST files in a directory; This update also fixes the problem that sst_dump exits with a success state even if target file/directory does not exist/is not an SST file/is empty/is corrupted. One test is added to sst_dump_test. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6943 Test Plan: Passed make check and a few manual tests Reviewed By: pdillinger Differential Revision: D21928985 Pulled By: gg814 fbshipit-source-id: 9a8b48e0cf1a0e96b13f42b690aba8ad981afad3	2020-06-08 13:58:28 -07:00
Zhichao Cao	fb08330f74	decouple the dependency of trace_analyzer_test unit test (#6941 ) Summary: Since gflags use the global variable to store the flags passed in. In the unit test, if we git one flag per unit test, the result is that all the flags are combined together in the following tests. Therefore, it has the dependency. In this PR, we pass the full arguments each time to ensure that the old arguments will be overwritten by the new one such that the dependency is removed. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6941 Test Plan: make asan_check. run each unit test in trace_analyzer_test independently and in arbitrary orders. Reviewed By: pdillinger Differential Revision: D21909176 Pulled By: zhichao-cao fbshipit-source-id: dca550a0a4a205c30faa620e258a020a3b5b4e13	2020-06-08 10:36:43 -07:00
Yanqin Jin	3020df9df5	Remove unnecessary inclusion of version_edit.h in env (#6952 ) Summary: In db_options.c, we should avoid including header files in the `db` directory to avoid introducing unnecessary dependency. The reason why `version_edit.h` has been included in `db_options.cc` is because we need two constants, `kUnknownChecksum` and `kUnknownChecksumFuncName`. We can put these two constants as `constexpr` in the public header `file_checksum.h`. Test plan (devserver): make check Pull Request resolved: https://github.com/facebook/rocksdb/pull/6952 Reviewed By: zhichao-cao Differential Revision: D21925341 Pulled By: riversand963 fbshipit-source-id: 2902f3b74c97f0cf16c58ad24c095c787c3a40e2	2020-06-07 21:56:55 -07:00
Levi Tamasi	f8c2e5a608	Do not print messages to stderr in VersionBuilder (#6948 ) Summary: RocksDB is an embedded library; we should not write to the application's console. Note: in each case, the same information is returned in the form of a `Status::Corruption` object. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6948 Test Plan: `make check` Reviewed By: ajkr Differential Revision: D21914965 Pulled By: ltamasi fbshipit-source-id: ae4b66789aa6b659eb8cc2ed4a048187962c86cc	2020-06-05 20:10:30 -07:00
Levi Tamasi	8988f831d8	Fix up a VersionBuilder test case (#6942 ) Summary: We currently do not have any validation that would ensure that the `FileMetaData` objects are equivalent when a file gets deleted from the LSM tree and then re-added (think trivial moves); however, if we did, this test case would be in violation. The patch changes the values used in the test case so they are consistent. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6942 Test Plan: `make check` Reviewed By: zhichao-cao Differential Revision: D21911366 Pulled By: ltamasi fbshipit-source-id: 2f0486f8337373a6a111b6f28433d70507857104	2020-06-05 18:16:38 -07:00
Zhichao Cao	f941adef88	Clean up the dead code (#6946 ) Summary: Remove the dead code in table test. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6946 Test Plan: run table_test Reviewed By: riversand963 Differential Revision: D21913563 Pulled By: zhichao-cao fbshipit-source-id: c0aa9f3b95dfe87dd7fb2cd4823784f08cb3ddd3	2020-06-05 17:45:22 -07:00
Zitan Chen	23e446a157	Disable OpenForReadOnly tests in the LITE mode (#6947 ) Summary: Disable two OpenForReadOnly tests in the LITE mode Pull Request resolved: https://github.com/facebook/rocksdb/pull/6947 Test Plan: passed db_test2 Reviewed By: cheng-chang Differential Revision: D21914345 Pulled By: gg814 fbshipit-source-id: 58e81baf5d8cf8adcedaef3966aa3a427bbdf7c2	2020-06-05 17:28:24 -07:00
sdong	2e7070b194	Directly use unit test tempalte buck (#6926 ) Summary: Make RocksDB run a predefined unit test so that it can be integrated with better tools. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6926 Test Plan: Watch tests Reviewed By: pdillinger Differential Revision: D21866216 fbshipit-source-id: cafca82efdf0b72671be8d30b665e88a75ae6000	2020-06-05 12:16:33 -07:00
anand76	98b0cbea88	Check iterator status BlockBasedTableReader::VerifyChecksumInBlocks() (#6909 ) Summary: The ```for``` loop in ```VerifyChecksumInBlocks``` only checks ```index_iter->Valid()``` which could be ```false``` either due to reaching the end of the index or, in case of partitioned index, it could be due to a checksum mismatch error when reading a 2nd level index block. Instead of throwing away the index iterator status, we need to return any errors back to the caller. Tests: Add a test in block_based_table_reader_test.cc. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6909 Reviewed By: pdillinger Differential Revision: D21833922 Pulled By: anand1976 fbshipit-source-id: bc778ebf1121dbbdd768689de5183f07a9f0beae	2020-06-05 11:08:25 -07:00
Cheng Chang	1bee0fca05	Make DestroyDir destroy directories recursively (#6934 ) Summary: Currently, `DeleteDir` only deletes the directory if there are no other directories under the target dir. This PR makes it delete directories recursively. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6934 Test Plan: Added a new unit test in testutil_test.cc. `make testutil_test` Reviewed By: zhichao-cao Differential Revision: D21884211 Pulled By: cheng-chang fbshipit-source-id: 0b9a48a200f494ee007aef5d1763b4aa331f8b5a	2020-06-05 10:55:22 -07:00
Akanksha Mahajan	2677bd5967	Add logs and stats in DeleteScheduler (#6927 ) Summary: Add logs and stats for files marked as trash and files deleted immediately in DeleteScheduler Pull Request resolved: https://github.com/facebook/rocksdb/pull/6927 Test Plan: make check -j64 Reviewed By: riversand963 Differential Revision: D21869068 Pulled By: akankshamahajan15 fbshipit-source-id: e9f673c4fa8049ce648b23c75d742f2f9c6c57a1	2020-06-05 09:43:04 -07:00