Add skip_memtable_flush in IngestExternalFileOptions

Summary: Set to true if you are sure although the key ranges may overlap, the memtable and the file being ingested do not have the same keys, or not care about that. The memtable flushing may block for hundreds of milliseconds. Test Plan: make check
2026-11-08 11:55:21 -05:00 · 2020-06-14 14:49:06 +08:00
78 changed files with 344 additions and 988 deletions
@@ -11,59 +11,7 @@ executors:
      shell: bash.exe

 jobs:
-  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:
+  build:
    executor: windows-2xlarge
    
    environment:
@@ -106,26 +54,3 @@ 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,9 +539,6 @@ 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
@@ -733,6 +730,9 @@ 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,7 +1090,6 @@ 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,7 +2,6 @@
 ## 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.
@@ -13,9 +12,6 @@
 * 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,7 +639,6 @@ TESTS = \
 	blob_file_garbage_test \
 	timer_test \
 	db_with_timestamp_compaction_test \
-	testutil_test \

 PARALLEL_TEST = \
 	backupable_db_test \
@@ -976,12 +975,6 @@ 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); \
@@ -995,10 +988,7 @@ check_0:
 	} \
 	  | $(prioritize_long_running_tests)				\
 	  | grep -E '$(tests-regexp)'					\
-	  | 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
+	  | build_tools/gnu_parallel -j$(J) --plain --joblog=LOG $$eta --gnu '{} >& t/log-{/}'

 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

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

 # TODO add ldb_tests
 check_some: $(SUBSET)
@@ -1878,9 +1866,6 @@ 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,6 +78,7 @@ 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 = {
@@ -124,9 +125,6 @@ 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",
@@ -321,6 +319,9 @@ 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",
@@ -397,10 +398,11 @@ cpp_library(
    os_deps = ROCKSDB_OS_DEPS,
    os_preprocessor_flags = ROCKSDB_OS_PREPROCESSOR_FLAGS,
    preprocessor_flags = ROCKSDB_PREPROCESSOR_FLAGS,
-    deps = [":rocksdb_lib"],
-    external_deps = ROCKSDB_EXTERNAL_DEPS + [
-        ("googletest", None, "gtest"),
+    deps = [
+        ":rocksdb_lib",
+        ":rocksdb_third_party_gtest",
    ],
+    external_deps = ROCKSDB_EXTERNAL_DEPS,
 )

 cpp_library(
@@ -447,6 +449,19 @@ 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"],
@@ -1470,13 +1485,6 @@ ROCKS_TESTS = [
        [],
        [],
    ],
-    [
-        "testutil_test",
-        "test_util/testutil_test.cc",
-        "serial",
-        [],
-        [],
-    ],
    [
        "thread_list_test",
        "util/thread_list_test.cc",
@@ -1630,17 +1638,18 @@ ROCKS_TESTS = [
 # Do not build the tests in opt mode, since SyncPoint and other test code
 # will not be included.
 [
-    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"),
-        ],
+    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,
    )
    for test_name, test_cc, parallelism, extra_deps, extra_compiler_flags in ROCKS_TESTS
    if not is_opt_mode
@@ -108,6 +108,16 @@ 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 = {
@@ -141,11 +151,13 @@ 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)
+    TARGETS = TARGETSBuilder("%s/TARGETS" % repo_path, gtest_dir)

    # rocksdb_lib
    TARGETS.add_library(
@@ -159,10 +171,7 @@ 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"],
-        extra_external_deps=""" + [
-        ("googletest", None, "gtest"),
-    ]""")
+        [":rocksdb_lib", ":rocksdb_third_party_gtest"])
    # rocksdb_tools_lib
    TARGETS.add_library(
        "rocksdb_tools_lib",
@@ -176,6 +185,12 @@ 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,10 +25,11 @@ def pretty_list(lst, indent=8):


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

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

    def add_rocksdb_library(self, name, srcs, headers=None):
@@ -85,13 +85,14 @@ 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")

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

@@ -182,17 +183,18 @@ ROCKS_TESTS = [
 # Do not build the tests in opt mode, since SyncPoint and other test code
 # will not be included.
 [
-    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"),
-        ],
+    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,
    )
    for test_name, test_cc, parallelism, extra_deps, extra_compiler_flags in ROCKS_TESTS
    if not is_opt_mode
@@ -239,9 +239,7 @@ Status BuildTable(
          (internal_stats == nullptr) ? nullptr
                                      : internal_stats->GetFileReadHist(0),
          TableReaderCaller::kFlush, /*arena=*/nullptr,
-          /*skip_filter=*/false, level,
-          MaxFileSizeForL0MetaPin(mutable_cf_options),
-          /*smallest_compaction_key=*/nullptr,
+          /*skip_filter=*/false, level, /*smallest_compaction_key=*/nullptr,
          /*largest_compaction_key*/ nullptr,
          /*allow_unprepared_value*/ false));
      s = it->status();
@@ -654,8 +654,6 @@ 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,6 +157,42 @@ 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;
@@ -175,7 +211,7 @@ class CorruptionTest : public testing::Test {
    }
    ASSERT_TRUE(!fname.empty()) << filetype;

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

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

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

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

 #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"
@@ -18,8 +17,6 @@
 #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 {

@@ -1343,57 +1340,6 @@ 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,15 +3184,6 @@ 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;
@@ -3223,7 +3214,6 @@ 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();
@@ -3238,16 +3228,6 @@ 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) {
@@ -4253,6 +4253,10 @@ Status DBImpl::IngestExternalFiles(
            "cannot ingest an external file into a dropped CF");
        break;
      }
+      if (args[i].options.skip_memtable_flush) {
+          need_flush[i] = false;
+          continue;
+      }
      bool tmp = false;
      status = ingestion_jobs[i].NeedsFlush(&tmp, cfd->GetSuperVersion());
      need_flush[i] = tmp;
@@ -1569,12 +1569,6 @@ 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,7 +376,6 @@ 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;
@@ -426,9 +425,6 @@ 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();
 }
@@ -820,12 +816,6 @@ 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,7 +25,6 @@ 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");
@@ -86,7 +85,6 @@ 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,8 +528,7 @@ class Repairer {
          cfd->GetLatestMutableCFOptions()->prefix_extractor.get(),
          /*table_reader_ptr=*/nullptr, /*file_read_hist=*/nullptr,
          TableReaderCaller::kRepair, /*arena=*/nullptr, /*skip_filters=*/false,
-          /*level=*/-1, /*max_file_size_for_l0_meta_pin=*/0,
-          /*smallest_compaction_key=*/nullptr,
+          /*level=*/-1, /*smallest_compaction_key=*/nullptr,
          /*largest_compaction_key=*/nullptr,
          /*allow_unprepared_value=*/false);
      ParsedInternalKey parsed;
@@ -97,8 +97,7 @@ 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,
-    size_t max_file_size_for_l0_meta_pin) {
+    bool prefetch_index_and_filter_in_cache) {
  std::string fname =
      TableFileName(ioptions_.cf_paths, fd.GetNumber(), fd.GetPathId());
  std::unique_ptr<FSRandomAccessFile> file;
@@ -125,8 +124,7 @@ 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_,
-                           max_file_size_for_l0_meta_pin),
+                           fd.largest_seqno, block_cache_tracer_),
        std::move(file_reader), fd.GetFileSize(), table_reader,
        prefetch_index_and_filter_in_cache);
    TEST_SYNC_POINT("TableCache::GetTableReader:0");
@@ -147,8 +145,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,
-                             size_t max_file_size_for_l0_meta_pin) {
+                             int level,
+                             bool prefetch_index_and_filter_in_cache) {
  PERF_TIMER_GUARD_WITH_ENV(find_table_nanos, ioptions_.env);
  Status s;
  uint64_t number = fd.GetNumber();
@@ -172,8 +170,7 @@ 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,
-                       max_file_size_for_l0_meta_pin);
+                       skip_filters, level, prefetch_index_and_filter_in_cache);
    if (!s.ok()) {
      assert(table_reader == nullptr);
      RecordTick(ioptions_.statistics, NO_FILE_ERRORS);
@@ -197,7 +194,6 @@ 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);
@@ -215,9 +211,7 @@ 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,
-                  true /* prefetch_index_and_filter_in_cache */,
-                  max_file_size_for_l0_meta_pin);
+                  skip_filters, level);
    if (s.ok()) {
      table_reader = GetTableReaderFromHandle(handle);
    }
@@ -378,7 +372,7 @@ Status TableCache::Get(const ReadOptions& options,
                       GetContext* get_context,
                       const SliceTransform* prefix_extractor,
                       HistogramImpl* file_read_hist, bool skip_filters,
-                       int level, size_t max_file_size_for_l0_meta_pin) {
+                       int level) {
  auto& fd = file_meta.fd;
  std::string* row_cache_entry = nullptr;
  bool done = false;
@@ -403,12 +397,10 @@ 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, true /* prefetch_index_and_filter_in_cache */,
-                    max_file_size_for_l0_meta_pin);
+      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);
      if (s.ok()) {
        t = GetTableReaderFromHandle(handle);
      }
@@ -72,8 +72,7 @@ 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, size_t max_file_size_for_l0_meta_pin,
-      const InternalKey* smallest_compaction_key,
+      bool skip_filters, int level, 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,
@@ -92,7 +91,7 @@ class TableCache {
             GetContext* get_context,
             const SliceTransform* prefix_extractor = nullptr,
             HistogramImpl* file_read_hist = nullptr, bool skip_filters = false,
-             int level = -1, size_t max_file_size_for_l0_meta_pin = 0);
+             int level = -1);

  // Return the range delete tombstone iterator of the file specified by
  // `file_meta`.
@@ -136,8 +135,7 @@ 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,
-                   size_t max_file_size_for_l0_meta_pin = 0);
+                   bool prefetch_index_and_filter_in_cache = true);

  // Get TableReader from a cache handle.
  TableReader* GetTableReaderFromHandle(Cache::Handle* handle);
@@ -202,8 +200,7 @@ 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,
-                        size_t max_file_size_for_l0_meta_pin = 0);
+                        bool prefetch_index_and_filter_in_cache = true);

  // 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,6 +278,7 @@ 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");
          }

@@ -286,6 +287,11 @@ 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) + " " +
@@ -295,6 +301,11 @@ 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) + " " +
@@ -309,6 +320,7 @@ 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");
          }
@@ -316,6 +328,9 @@ 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. " +
@@ -754,8 +769,7 @@ 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,
-                           size_t max_file_size_for_l0_meta_pin) {
+                           const SliceTransform* prefix_extractor) {
    assert(table_cache_ != nullptr);

    size_t table_cache_capacity = table_cache_->get_cache()->GetCapacity();
@@ -824,7 +838,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, max_file_size_for_l0_meta_pin);
+            prefetch_index_and_filter_in_cache);
        if (file_meta->table_reader_handle != nullptr) {
          // Load table_reader
          file_meta->fd.table_reader = table_cache_->GetTableReaderFromHandle(
@@ -883,11 +897,10 @@ 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,
-    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);
+    const SliceTransform* prefix_extractor) {
+  return rep_->LoadTableHandlers(internal_stats, max_threads,
+                                 prefetch_index_and_filter_in_cache,
+                                 is_initial_load, prefix_extractor);
 }

 BaseReferencedVersionBuilder::BaseReferencedVersionBuilder(
@@ -42,8 +42,7 @@ 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,
-                           size_t max_file_size_for_l0_meta_pin);
+                           const SliceTransform* prefix_extractor);

 private:
  class Rep;
@@ -422,19 +422,8 @@ 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, file_size, path_id, smallest_seq,
-      largest_seq, num_entries, num_deletions, sampled, smallest_seqno,
-      largest_seqno);
+  Add(level, file_number, smallest, largest);

  EnvOptions env_options;
  constexpr TableCache* table_cache = nullptr;
@@ -451,14 +440,18 @@ 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, smallest_seq),
-                   GetInternalKey(largest, largest_seq), smallest_seqno,
-                   largest_seqno, marked_for_compaction, kInvalidBlobFileNumber,
-                   kUnknownOldestAncesterTime, kUnknownFileCreationTime,
-                   kUnknownFileChecksum, kUnknownFileChecksumFuncName);
+                   GetInternalKey(smallest), GetInternalKey(largest),
+                   smallest_seqno, largest_seqno, marked_for_compaction,
+                   kInvalidBlobFileNumber, kUnknownOldestAncesterTime,
+                   kUnknownFileCreationTime, kUnknownFileChecksum,
+                   kUnknownFileChecksumFuncName);

  ASSERT_OK(builder.Apply(&addition));

@@ -18,6 +18,10 @@
 #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,6 +29,9 @@ 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,8 +414,7 @@ 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(),
-      MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
+      cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
  if ((s.IsPathNotFound() || s.IsCorruption()) &&
      no_error_if_table_files_missing_) {
    s = Status::OK();
@@ -365,7 +365,6 @@ 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()),
@@ -482,19 +481,9 @@ 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. 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_;
-        }
+        // 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
        break;
      } else {
        if (curr_level_ == 0) {
@@ -514,12 +503,6 @@ 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;
  }

@@ -541,7 +524,7 @@ class FilePickerMultiGet {
          // file regardless for all keys not found yet
          if (current_level_range_.CheckKeyDone(batch_iter_) ||
              curr_level_ == 0) {
-            batch_iter_ = upper_key_;
+            ++batch_iter_;
          }
        }
        // batch_iter_prev_ will become the start key for the next file
@@ -561,20 +544,18 @@ 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
-          // 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;
-          }
+          // 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);
          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_ =
@@ -626,7 +607,6 @@ 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_;
@@ -646,7 +626,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();
-        upper_key_ = batch_iter_ = current_level_range_.begin();
+        batch_iter_ = current_level_range_.begin();
        return true;
      }
    }
@@ -741,7 +721,7 @@ class FilePickerMultiGet {
      }
      if (level_contains_keys) {
        batch_iter_prev_ = current_level_range_.begin();
-        upper_key_ = batch_iter_ = current_level_range_.begin();
+        batch_iter_ = current_level_range_.begin();
        return true;
      }
      curr_level_++;
@@ -1004,8 +984,7 @@ 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_,
-        /*max_file_size_for_l0_meta_pin=*/0, smallest_compaction_key,
+        /*arena=*/nullptr, skip_filters_, level_, smallest_compaction_key,
        largest_compaction_key, allow_unprepared_value_);
  }

@@ -1600,9 +1579,10 @@ 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, max_file_size_for_l0_meta_pin_,
+          /*skip_filters=*/false, /*level=*/0,
          /*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
@@ -1624,8 +1604,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,
-        /*compaction_boundaries=*/nullptr, allow_unprepared_value));
+        range_del_agg, /*largest_compaction_key=*/nullptr,
+        allow_unprepared_value));
  }
 }

@@ -1659,7 +1639,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, max_file_size_for_l0_meta_pin_,
+          /*skip_filters=*/false, /*level=*/0,
          /*smallest_compaction_key=*/nullptr,
          /*largest_compaction_key=*/nullptr,
          /*allow_unprepared_value=*/false));
@@ -1763,8 +1743,6 @@ 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,
@@ -1828,7 +1806,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(), max_file_size_for_l0_meta_pin_);
+        fp.GetHitFileLevel());
    // TODO: examine the behavior for corrupted key
    if (timer_enabled) {
      PERF_COUNTER_BY_LEVEL_ADD(get_from_table_nanos, timer.ElapsedNanos(),
@@ -3924,8 +3902,7 @@ 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(),
-            MaxFileSizeForL0MetaPin(*mutable_cf_options_ptrs[i]));
+            mutable_cf_options_ptrs[i]->prefix_extractor.get());
        if (!s.ok()) {
          if (db_options_->paranoid_checks) {
            break;
@@ -4652,8 +4629,7 @@ 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(),
-          MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
+          cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
      if (!s.ok()) {
        if (db_options_->paranoid_checks) {
          return s;
@@ -5690,14 +5666,13 @@ 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)),
-              MaxFileSizeForL0MetaPin(*c->mutable_cf_options()),
+              /*skip_filters=*/false, /*level=*/static_cast<int>(c->level(which)),
              /*smallest_compaction_key=*/nullptr,
              /*largest_compaction_key=*/nullptr,
              /*allow_unprepared_value=*/false);
@@ -6028,8 +6003,7 @@ 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(),
-                MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
+                cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
            if (!s.ok()) {
              enough = false;
              if (s.IsPathNotFound()) {
@@ -6305,8 +6279,7 @@ 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(),
-        MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
+        cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
    TEST_SYNC_POINT_CALLBACK(
        "ReactiveVersionSet::ApplyOneVersionEditToBuilder:"
        "AfterLoadTableHandlers",
@@ -324,17 +324,6 @@ 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_; }
@@ -803,8 +792,6 @@ 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, FileLocationAndMetaDataByNumber) {
+TEST_F(VersionStorageInfoTest, FileLocation) {
  Add(0, 11U, "1", "2", 5000U);
  Add(0, 12U, "1", "2", 5000U);

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

  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,7 +7,6 @@

 #include "file/delete_scheduler.h"

-#include <cinttypes>
 #include <thread>
 #include <vector>

@@ -66,12 +65,6 @@ 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;
  }
@@ -79,8 +72,6 @@ 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",
@@ -88,14 +79,10 @@ 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);
@@ -223,8 +210,6 @@ 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
@@ -248,27 +233,19 @@ void DeleteScheduler::BackgroundEmptyTrash() {
        bg_errors_[path_in_trash] = s;
      }

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

      if (is_complete) {
        pending_files_--;
@@ -376,13 +353,9 @@ 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,10 +83,6 @@ 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);

@@ -141,7 +137,6 @@ 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,7 +32,6 @@ class DeleteSchedulerTest : public testing::Test {
          ToString(i));
      DestroyAndCreateDir(dummy_files_dirs_.back());
    }
-    stats_ = ROCKSDB_NAMESPACE::CreateDBStatistics();
  }

  ~DeleteSchedulerTest() override {
@@ -100,7 +99,6 @@ 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_;
@@ -108,7 +106,6 @@ 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).
@@ -185,8 +182,6 @@ 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();
  }
 }
@@ -224,9 +219,6 @@ 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();
 }

@@ -309,10 +301,6 @@ 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();
  }
 }
@@ -330,9 +318,8 @@ TEST_F(DeleteSchedulerTest, DisableRateLimiting) {

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

-  for (int i = 0; i < num_files; i++) {
+  for (int i = 0; i < 10; i++) {
    // Every file we delete will be deleted immediately
    std::string dummy_file = NewDummyFile("dummy.data");
    ASSERT_OK(delete_scheduler_->DeleteFile(dummy_file, ""));
@@ -342,9 +329,6 @@ 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();
 }
@@ -381,8 +365,6 @@ 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();
 }
@@ -457,12 +439,9 @@ 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();
 }

@@ -674,8 +653,6 @@ 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,11 +135,6 @@ 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,
@@ -195,7 +190,6 @@ 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;
+    return kUnknownFileChecksumFuncName.c_str();
  }
 }

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

 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;
 };
@@ -1571,6 +1571,11 @@ struct IngestExternalFileOptions {
  // Using a large readahead size (> 2MB) can typically improve the performance
  // of forward iteration on spinning disks.
  size_t verify_checksums_readahead_size = 0;
+  // Set to true if you are sure although the key ranges may overlap, 
+  // the memtable and the file being ingested
+  // do not have the same keys, or not care about that.
+  // The memtable flushing may block for hundreds of milliseconds.
+  bool skip_memtable_flush = false;
 };

 enum TraceFilterType : uint64_t {
@@ -11,7 +11,7 @@ namespace ROCKSDB_NAMESPACE {

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

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

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

 namespace ROCKSDB_NAMESPACE {
@@ -81,9 +80,6 @@ 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,12 +360,6 @@ 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 const* const* argv, const Options& options,
+      int argc, char** argv, const Options& options,
      const LDBOptions& ldb_options,
      const std::vector<ColumnFamilyDescriptor>* column_families);

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

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

  // Returns the status code to return. 0 is no error.
  static int RunCommand(
-      int argc, char const* const* argv, Options options,
-      const LDBOptions& ldb_options,
+      int argc, char** 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,11 +4945,6 @@ 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;
@@ -5245,10 +5240,6 @@ 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,16 +722,6 @@ 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,8 +184,6 @@ 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,17 +720,6 @@ 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,9 +268,4 @@ 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,6 +7,7 @@

 #include <cinttypes>

+#include "db/version_edit.h"
 #include "logging/logging.h"
 #include "options/options_helper.h"
 #include "port/port.h"
@@ -615,8 +616,9 @@ 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);
+                   file_checksum_gen_factory
+                       ? file_checksum_gen_factory->Name()
+                       : kUnknownFileChecksumFuncName.c_str());
  ROCKS_LOG_HEADER(log, "                Options.best_efforts_recovery: %d",
                   static_cast<int>(best_efforts_recovery));
 }
@@ -8,9 +8,6 @@ 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                                           \
@@ -201,6 +198,9 @@ 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;
+    return kUnknownFileChecksumFuncName.c_str();
  }
 }

@@ -420,8 +420,7 @@ 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.max_file_size_for_l0_meta_pin);
+      table_reader_options.block_cache_tracer);
 }

 TableBuilder* BlockBasedTableFactory::NewTableBuilder(
@@ -610,8 +610,7 @@ 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,
-    size_t max_file_size_for_l0_meta_pin) {
+    BlockCacheTracer* const block_cache_tracer) {
  table_reader->reset();

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

  if (s.ok()) {
    // Update tail prefetch stats
@@ -945,7 +943,6 @@ 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;

@@ -990,10 +987,9 @@ 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 && file_size <= max_file_size_for_l0_meta_pin;
+      rep_->table_options.pin_l0_filter_and_index_blocks_in_cache && level == 0;

  // prefetch the first level of index
  const bool prefetch_index =
@@ -2841,12 +2837,6 @@ 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,8 +101,7 @@ 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,
-                     size_t max_file_size_for_l0_meta_pin = 0);
+                     BlockCacheTracer* const block_cache_tracer = nullptr);

  bool PrefixMayMatch(const Slice& internal_key,
                      const ReadOptions& read_options,
@@ -256,7 +255,6 @@ 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_;

@@ -422,7 +420,6 @@ 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,8 +4,6 @@
 //  (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"
@@ -21,29 +19,19 @@ namespace ROCKSDB_NAMESPACE {

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

  void SetUp() override {
-    BlockBasedTableOptions::IndexType index_type;
-    bool no_block_cache;
-    std::tie(compression_type_, use_direct_reads_, index_type, no_block_cache) =
-        GetParam();
+    std::tie(compression_type_, use_direct_reads_) = 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_)); }
@@ -62,7 +50,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 */,
@@ -91,21 +79,19 @@ 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_;
-  std::unique_ptr<BlockBasedTableFactory> table_factory_;
+  BlockBasedTableFactory table_factory_;
+
+  std::string Path(const std::string& fname) { return test_dir_ + "/" + fname; }

  void WriteToFile(const std::string& content, const std::string& filename) {
    std::unique_ptr<FSWritableFile> f;
@@ -233,104 +219,20 @@ 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::Values(BlockBasedTableOptions::IndexType::kBinarySearch),
-        ::testing::Values(false)));
+    ::testing::Combine(::testing::ValuesIn(GetSupportedCompressions()),
+                       ::testing::Values(false)));
 #else   // ROCKSDB_LITE
 INSTANTIATE_TEST_CASE_P(
    MultiGet, BlockBasedTableReaderTest,
-    ::testing::Combine(
-        ::testing::ValuesIn(GetSupportedCompressions()), ::testing::Bool(),
-        ::testing::Values(BlockBasedTableOptions::IndexType::kBinarySearch),
-        ::testing::Values(false)));
+    ::testing::Combine(::testing::ValuesIn(GetSupportedCompressions()),
+                       ::testing::Bool()));
 #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,7 +122,6 @@ 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;
+    return kUnknownFileChecksumFuncName.c_str();
  }
 }

@@ -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;
+    return kUnknownFileChecksumFuncName.c_str();
  }

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

@@ -34,13 +34,11 @@ 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,
-                     size_t _max_file_size_for_l0_meta_pin = 0)
+                     BlockCacheTracer* const _block_cache_tracer = nullptr)
      : TableReaderOptions(_ioptions, _prefix_extractor, _env_options,
                           _internal_comparator, _skip_filters, _immortal,
                           _force_direct_prefetch, _level,
-                           0 /* _largest_seqno */, _block_cache_tracer,
-                           _max_file_size_for_l0_meta_pin) {}
+                           0 /* _largest_seqno */, _block_cache_tracer) {}

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

  const ImmutableCFOptions& ioptions;
  const SliceTransform* prefix_extractor;
@@ -81,9 +77,6 @@ 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,8 +378,7 @@ 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_,
-                           moptions.write_buffer_size),
+                           false, level_, largest_seqno_, &block_cache_tracer_),
        std::move(file_reader_), TEST_GetSink()->contents().size(),
        &table_reader_);
  }
@@ -3266,9 +3265,17 @@ 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);
@@ -3285,8 +3292,9 @@ TEST_P(BlockBasedTableTest, NoFileChecksum) {
  f.ResetTableBuilder(std::move(builder));
  f.AddKVtoKVMap(1000);
  f.WriteKVAndFlushTable();
-  ASSERT_STREQ(f.GetFileChecksumFuncName(), kUnknownFileChecksumFuncName);
-  ASSERT_STREQ(f.GetFileChecksum().c_str(), kUnknownFileChecksum);
+  ASSERT_STREQ(f.GetFileChecksumFuncName(),
+               kUnknownFileChecksumFuncName.c_str());
+  ASSERT_STREQ(f.GetFileChecksum().c_str(), kUnknownFileChecksum.c_str());
 }

 TEST_P(BlockBasedTableTest, Crc32cFileChecksum) {
@@ -3299,9 +3307,17 @@ 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;
@@ -3430,8 +3446,9 @@ TEST_F(PlainTableTest, NoFileChecksum) {
  f.ResetTableBuilder(std::move(builder));
  f.AddKVtoKVMap(1000);
  f.WriteKVAndFlushTable();
-  ASSERT_STREQ(f.GetFileChecksumFuncName(), kUnknownFileChecksumFuncName);
-  EXPECT_EQ(f.GetFileChecksum(), kUnknownFileChecksum);
+  ASSERT_STREQ(f.GetFileChecksumFuncName(),
+               kUnknownFileChecksumFuncName.c_str());
+  EXPECT_EQ(f.GetFileChecksum(), kUnknownFileChecksum.c_str());
 }

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

 #include <fcntl.h>
-#include <sys/stat.h>
 #include <array>
 #include <cctype>
 #include <fstream>
@@ -465,16 +464,7 @@ Status DestroyDir(Env* env, const std::string& dir) {
      if (file_in_dir == "." || file_in_dir == "..") {
        continue;
      }
-      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);
-        }
-      }
+      s = env->DeleteFile(dir + "/" + file_in_dir);
      if (!s.ok()) {
        break;
      }
@@ -537,46 +527,5 @@ 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,7 +812,5 @@ 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
@@ -1,42 +0,0 @@
-//  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 const* const* argv, const Options& options,
+    int argc, char** argv, const Options& options,
    const LDBOptions& ldb_options,
    const std::vector<ColumnFamilyDescriptor>* column_families) {
  std::vector<std::string> args;
@@ -41,18 +41,6 @@ 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*/, bool to_stderr) {
+                                 const char* /*exec_name*/) {
  std::string ret;

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

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

 int LDBCommandRunner::RunCommand(
-    int argc, char const* const* argv, Options options,
-    const LDBOptions& ldb_options,
+    int argc, char** argv, Options options, const LDBOptions& ldb_options,
    const std::vector<ColumnFamilyDescriptor>* column_families) {
  if (argc <= 2) {
-    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;
-    }
+    PrintHelp(ldb_options, argv[0]);
+    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], /*to_stderr*/ true);
+    PrintHelp(ldb_options, argv[0]);
    return 1;
  }

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

 int main(int argc, char** argv) {
  ROCKSDB_NAMESPACE::SSTDumpTool tool;
-  return tool.Run(argc, argv);
+  tool.Run(argc, argv);
+  return 0;
 }
 #else
 #include <stdio.h>
@@ -129,20 +129,6 @@ 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();
@@ -312,50 +298,13 @@ 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));
-  }
-  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,12 +95,6 @@ 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));

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

 namespace {

-void print_help(bool to_stderr) {
+void print_help() {
  fprintf(
-      to_stderr ? stderr : stdout,
-      R"(sst_dump --file=<data_dir_OR_sst_file> [--command=check|scan|raw|recompress|identify]
+      stderr,
+      R"(sst_dump --file=<data_dir_OR_sst_file> [--command=check|scan|raw|recompress]
    --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|identify
+    --command=check|scan|raw|verify
        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
@@ -527,7 +520,7 @@ void print_help(bool to_stderr) {

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

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

-int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
+int SSTDumpTool::Run(int argc, char** argv, Options options) {
  const char* env_uri = nullptr;
  const char* dir_or_file = nullptr;
  uint64_t read_num = std::numeric_limits<uint64_t>::max();
@@ -696,17 +689,10 @@ int SSTDumpTool::Run(int argc, char const* const* 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(/*to_stderr*/ true);
-      return 1;
+      print_help();
+      exit(1);
    }
  }

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

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

@@ -762,29 +748,17 @@ int SSTDumpTool::Run(int argc, char const* const* 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() || 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)
+  if (!st.ok()) {
    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 ||
@@ -792,7 +766,6 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
      // ignore
      continue;
    }
-
    if (dir) {
      filename = std::string(dir_or_file) + "/" + filename;
    }
@@ -800,23 +773,10 @@ int SSTDumpTool::Run(int argc, char const* const* 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") {
@@ -921,35 +881,7 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
    fprintf(stdout, "total filter block size: %" PRIu64 "\n",
            total_filter_block_size);
  }
-
-  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;
-  }
+  return 0;
 }
 }  // namespace ROCKSDB_NAMESPACE

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

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

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

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

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

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

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

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

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

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

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

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

@@ -19,9 +19,6 @@
 #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"
@@ -33,6 +30,9 @@
 #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,11 +8,10 @@
 #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 {
@@ -5,7 +5,7 @@
 //
 #ifndef ROCKSDB_LITE

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

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

-#include "db/blob/blob_log_reader.h"
+#include "utilities/blob_db/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 "db/blob/blob_log_writer.h"
+#include "utilities/blob_db/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 {