rust-rocksdb 0.50.0

// Copyright 2020 Tyler Neely
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

mod util;

use std::{
    fs,
    io::Read as _,
    sync::{Arc, LazyLock, atomic::AtomicUsize},
};

use rust_rocksdb::{
    BlockBasedOptions, BlockBasedPinningTier, Cache, DB, DBCompactionPri, DBCompressionType,
    DataBlockIndexType, Env, IndexBlockSearchType, LruCacheOptions, Options, ReadOptions,
    checkpoint::Checkpoint,
};
use util::DBPath;

#[test]
fn test_load_latest() {
    let n = DBPath::new("_rust_rocksdb_test_load_latest");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.create_missing_column_families(true);
        let _ = DB::open_cf(&opts, &n, vec!["cf0", "cf1"]).unwrap();
    }
    let (_, cfs) = Options::load_latest(
        &n,
        Env::new().unwrap(),
        true,
        Cache::new_lru_cache(1024 * 8),
    )
    .unwrap();
    assert!(cfs.iter().any(|cf| cf.name() == "default"));
    assert!(cfs.iter().any(|cf| cf.name() == "cf0"));
    assert!(cfs.iter().any(|cf| cf.name() == "cf1"));
}
#[test]
fn test_add_compact_on_deletion_collector_factory_min_file_size() {
    let n =
        DBPath::new("_rust_rocksdb_test_add_compact_on_deletion_collector_factory_min_file_size");

    let mut opts = Options::default();
    opts.create_if_missing(true);
    opts.add_compact_on_deletion_collector_factory_min_file_size(5, 10, 0.5, 32 * 1024);
    let _db = DB::open(&opts, &n).unwrap();

    // We can't easily assert the min_file_size in LOG (not printed),
    // but the factory ToString is included and should match.
    let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
        .expect("rocksdb creates a LOG file");
    let mut settings = String::new();
    rocksdb_log
        .read_to_string(&mut settings)
        .expect("can read the LOG file");
    assert!(settings.contains("CompactOnDeletionCollector (Sliding window size = 5 Deletion trigger = 10 Deletion ratio = 0.5)"));
}

#[test]
fn test_get_options_from_string() {
    let mut opts = Options::default();
    opts.set_use_fsync(true);
    let new_opts = opts.get_options_from_string("use_fsync=false").unwrap();
    assert!(!new_opts.get_use_fsync());
    assert!(
        opts.get_options_from_string("notarealoptionstring")
            .is_err()
    );

    // test an option with a NUL byte
    let err = match opts.get_options_from_string("use_fsync=false\0") {
        Ok(_) => panic!("expected error"),
        Err(e) => e,
    };
    assert!(
        err.to_string().contains("must not contain NUL"),
        "err={err:?}"
    );
}

#[test]
fn test_set_num_levels() {
    let n = DBPath::new("_rust_rocksdb_test_set_num_levels");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_num_levels(2);
        let _db = DB::open(&opts, &n).unwrap();
    }
}

#[test]
fn test_increase_parallelism() {
    let n = DBPath::new("_rust_rocksdb_test_increase_parallelism");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.increase_parallelism(4);
        let _db = DB::open(&opts, &n).unwrap();
    }
}

#[test]
fn test_set_level_compaction_dynamic_level_bytes() {
    let n = DBPath::new("_rust_rocksdb_test_set_level_compaction_dynamic_level_bytes");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_level_compaction_dynamic_level_bytes(true);
        let _db = DB::open(&opts, &n).unwrap();
    }
}

#[test]
fn test_block_based_options() {
    let path = "_rust_rocksdb_test_block_based_options";
    let n = DBPath::new(path);
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);

        let mut block_opts = BlockBasedOptions::default();
        block_opts.set_cache_index_and_filter_blocks(true);
        block_opts.set_pin_l0_filter_and_index_blocks_in_cache(true);
        block_opts.set_format_version(4);
        block_opts.set_index_block_restart_interval(16);
        block_opts.set_top_level_index_pinning_tier(BlockBasedPinningTier::All);
        block_opts.set_partition_pinning_tier(BlockBasedPinningTier::All);
        block_opts.set_unpartitioned_pinning_tier(BlockBasedPinningTier::All);

        opts.set_block_based_table_factory(&block_opts);
        let _db = DB::open(&opts, &n).unwrap();

        // read the setting from the LOG file
        let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
            .expect("rocksdb creates a LOG file");
        let mut settings = String::new();
        rocksdb_log.read_to_string(&mut settings).unwrap();

        // check the settings are set in the LOG file
        assert!(settings.contains("cache_index_and_filter_blocks: 1"));
        assert!(settings.contains("pin_l0_filter_and_index_blocks_in_cache: 1"));
        assert!(settings.contains("format_version: 4"));
        assert!(settings.contains("index_block_restart_interval: 16"));
    }

    {
        let mut opts = Options::default();
        opts.create_if_missing(true);

        let mut block_opts = BlockBasedOptions::default();
        block_opts.set_cache_index_and_filter_blocks(true);
        block_opts.set_top_level_index_pinning_tier(BlockBasedPinningTier::FlushAndSimilar);
        block_opts.set_partition_pinning_tier(BlockBasedPinningTier::All);
        block_opts.set_unpartitioned_pinning_tier(BlockBasedPinningTier::None);

        opts.set_block_based_table_factory(&block_opts);
        let _db = DB::open(&opts, &n).unwrap();
    }
}

#[test]
fn test_read_options() {
    let mut read_opts = ReadOptions::default();
    read_opts.set_verify_checksums(false);
    read_opts.set_deadline(121);
    read_opts.set_io_timeout(343);

    // `optimize_multiget_for_io` is a ReadOptions flag consulted inside
    // `USE_COROUTINES` builds only. It is not persisted to the DB's LOG.
    // Round-trip through the getter to confirm the C-API setter actually
    // mutates the underlying ReadOptions struct rather than being a no-op
    // shim. C++ default is `true`.
    assert!(read_opts.get_optimize_multiget_for_io());
    read_opts.set_optimize_multiget_for_io(false);
    assert!(!read_opts.get_optimize_multiget_for_io());
    read_opts.set_optimize_multiget_for_io(true);
    assert!(read_opts.get_optimize_multiget_for_io());
}

#[test]
fn test_set_data_block_index_type() {
    let path = "_rust_rocksdb_test_set_data_block_index_type";
    let n = DBPath::new(path);

    // Default is `BinarySearch`
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);

        let block_opts = BlockBasedOptions::default();
        opts.set_block_based_table_factory(&block_opts);
        let _db = DB::open(&opts, &n).expect("open a db works");

        let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
            .expect("rocksdb creates a LOG file");
        let mut settings = String::new();
        rocksdb_log
            .read_to_string(&mut settings)
            .expect("can read the LOG file");
        assert!(settings.contains("data_block_index_type: 0"));
        assert!(settings.contains("data_block_hash_table_util_ratio: 0.750000"));
    }

    // Setting the index type and hash table utilization ratio works
    {
        let mut opts = Options::default();
        opts.create_if_missing(false);

        let mut block_opts = BlockBasedOptions::default();
        block_opts.set_data_block_index_type(DataBlockIndexType::BinaryAndHash);
        block_opts.set_data_block_hash_ratio(0.35);
        opts.set_block_based_table_factory(&block_opts);
        let _db = DB::open(&opts, &n).expect("open a db works");

        let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
            .expect("rocksdb creates a LOG file");
        let mut settings = String::new();
        rocksdb_log
            .read_to_string(&mut settings)
            .expect("can read the LOG file");
        assert!(settings.contains("data_block_index_type: 1"));
        assert!(settings.contains("data_block_hash_table_util_ratio: 0.350000"));
    }
}

#[test]
fn test_set_index_block_search_type_and_uniform_cv_threshold() {
    let path = "_rust_rocksdb_test_index_block_search_type";
    let n = DBPath::new(path);

    // Default state: uniform_cv_threshold should be the disabled sentinel
    // (-1) so that BlockSearchType::Auto degenerates to binary search.
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);

        let block_opts = BlockBasedOptions::default();
        opts.set_block_based_table_factory(&block_opts);
        let _db = DB::open(&opts, &n).expect("open default-options db works");

        let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
            .expect("rocksdb creates a LOG file");
        let mut settings = String::new();
        rocksdb_log
            .read_to_string(&mut settings)
            .expect("can read the LOG file");
        // RocksDB prints `uniform_cv_threshold` regardless of value; the
        // -1.0 default lands in the LOG as `-1.000000`.
        assert!(
            settings.contains("uniform_cv_threshold: -1.000000"),
            "expected default uniform_cv_threshold in LOG; got:\n{settings}",
        );
    }

    // Enable Auto + a real threshold and verify the threshold round-trips
    // into the LOG. The search-type field itself is not printed by
    // BlockBasedTableFactory::GetPrintableOptions today, so we cannot read
    // it back; the threshold is the meaningful behavioural input for Auto
    // and is sufficient to confirm the wiring.
    {
        let mut opts = Options::default();
        opts.create_if_missing(false);

        let mut block_opts = BlockBasedOptions::default();
        block_opts.set_index_block_search_type(IndexBlockSearchType::Auto);
        block_opts.set_uniform_cv_threshold(0.2);
        opts.set_block_based_table_factory(&block_opts);
        let _db = DB::open(&opts, &n).expect("open db with Auto + threshold works");

        let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
            .expect("rocksdb creates a LOG file");
        let mut settings = String::new();
        rocksdb_log
            .read_to_string(&mut settings)
            .expect("can read the LOG file");
        assert!(
            settings.contains("uniform_cv_threshold: 0.200000"),
            "expected configured uniform_cv_threshold in LOG; got:\n{settings}",
        );
    }
}

#[test]
fn test_set_index_block_search_type_all_variants() {
    // Each variant in turn must round-trip through the setter without
    // panicking and the resulting DB must open. This guards against drift
    // between the Rust enum discriminants and the C API enum constants.
    //
    // We use an explicit (variant, name) table instead of `format!("{:?}")`
    // so the on-disk path doesn't silently change if a future
    // `#[derive(Debug)]` is replaced with a custom `impl Debug`.
    let cases: &[(IndexBlockSearchType, &str)] = &[
        (IndexBlockSearchType::Binary, "binary"),
        (IndexBlockSearchType::Interpolation, "interpolation"),
        (IndexBlockSearchType::Auto, "auto"),
    ];
    for (variant, name) in cases {
        let n = DBPath::new(&format!(
            "_rust_rocksdb_test_index_block_search_type_variant_{name}"
        ));
        let mut opts = Options::default();
        opts.create_if_missing(true);

        let mut block_opts = BlockBasedOptions::default();
        block_opts.set_index_block_search_type(*variant);
        // Set a valid threshold so Auto has something meaningful to read,
        // and so we cover the case where the threshold is set even with
        // Binary/Interpolation (which simply ignore it at read time).
        block_opts.set_uniform_cv_threshold(0.2);
        opts.set_block_based_table_factory(&block_opts);

        let _db = DB::open(&opts, &n).unwrap_or_else(|e| panic!("open db with {name} failed: {e}"));
    }
}

#[test]
fn test_set_index_block_search_type_auto_with_default_threshold_degenerates() {
    // The documented contract of `IndexBlockSearchType::Auto` is that it
    // selects per-block search using the write-path `is_uniform` footer bit
    // — which is never set when `uniform_cv_threshold` is left at its
    // disabled-sentinel default of `-1`. In that combination `Auto` is
    // documented to "degenerate to binary search".
    //
    // We can't observe the lookup path from Rust, but we CAN pin down the
    // weaker (and the only externally testable) guarantee: opening a DB
    // with `Auto` and NO call to `set_uniform_cv_threshold` must succeed
    // and must report the default `-1.000000` threshold in the LOG. This
    // catches a future regression where e.g. the default sentinel was
    // changed and `Auto` started failing at DB-open instead of falling
    // back silently.
    let n = DBPath::new("_rust_rocksdb_test_index_block_search_type_auto_default_threshold");
    let mut opts = Options::default();
    opts.create_if_missing(true);

    let mut block_opts = BlockBasedOptions::default();
    block_opts.set_index_block_search_type(IndexBlockSearchType::Auto);
    // Intentionally NOT calling set_uniform_cv_threshold — exercise the
    // documented degenerate path.
    opts.set_block_based_table_factory(&block_opts);

    let _db = DB::open(&opts, &n).expect("open with Auto + default threshold must succeed");

    let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
        .expect("rocksdb creates a LOG file");
    let mut settings = String::new();
    rocksdb_log
        .read_to_string(&mut settings)
        .expect("can read the LOG file");
    assert!(
        settings.contains("uniform_cv_threshold: -1.000000"),
        "expected default uniform_cv_threshold (-1) in LOG; got:\n{settings}",
    );
}

#[test]
fn test_set_memtable_batch_lookup_optimization() {
    let path = "_rust_rocksdb_test_memtable_batch_lookup_optimization";
    let n = DBPath::new(path);

    // Round-trip the setter through the getter first, before any DB open.
    // This confirms the C-API setter is actually wired through to the
    // ColumnFamilyOptions field, independently of whatever the LOG dumper
    // emits.
    let mut opts = Options::default();
    assert!(
        !opts.get_memtable_batch_lookup_optimization(),
        "default C++ value is false",
    );
    opts.set_memtable_batch_lookup_optimization(true);
    assert!(opts.get_memtable_batch_lookup_optimization());
    opts.set_memtable_batch_lookup_optimization(false);
    assert!(!opts.get_memtable_batch_lookup_optimization());

    // Default: disabled. The cf_options dumper prefixes immutable
    // ColumnFamilyOptions fields with `Options.` in the LOG.
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        let _db = DB::open(&opts, &n).expect("open default-options db works");

        let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
            .expect("rocksdb creates a LOG file");
        let mut settings = String::new();
        rocksdb_log
            .read_to_string(&mut settings)
            .expect("can read the LOG file");
        assert!(
            settings.contains("Options.memtable_batch_lookup_optimization: false"),
            "expected default memtable_batch_lookup_optimization=false in LOG; got:\n{settings}",
        );
    }

    // Enabled.
    {
        let mut opts = Options::default();
        opts.create_if_missing(false);
        opts.set_memtable_batch_lookup_optimization(true);
        let _db = DB::open(&opts, &n).expect("open db with batch lookup opt works");

        let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
            .expect("rocksdb creates a LOG file");
        let mut settings = String::new();
        rocksdb_log
            .read_to_string(&mut settings)
            .expect("can read the LOG file");
        assert!(
            settings.contains("Options.memtable_batch_lookup_optimization: true"),
            "expected configured memtable_batch_lookup_optimization=true in LOG; \
             got:\n{settings}",
        );
    }
}

#[test]
#[cfg(feature = "zstd")]
fn set_compression_options_zstd_max_train_bytes() {
    let path = DBPath::new("_rust_set_compression_options_zstd_max_train_bytes");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_compression_options(4, 5, 6, 7);
        opts.set_zstd_max_train_bytes(100);
        let _db = DB::open(&opts, &path).unwrap();
    }
}

#[test]
fn set_wal_compression_zstd() {
    let path = DBPath::new("_set_wal_compression_zstd");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_wal_compression_type(DBCompressionType::None);
        opts.set_wal_compression_type(DBCompressionType::Zstd);
        let _db = DB::open(&opts, &path).unwrap();
    }
}

#[test]
#[should_panic(expected = "Lz4 is not supported for WAL compression")]
fn set_wal_compression_unsupported() {
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_wal_compression_type(DBCompressionType::Lz4);
    }
}

fn test_compression_type(ty: DBCompressionType) {
    let path = DBPath::new("_test_compression_type");

    let mut opts = Options::default();
    opts.set_compression_type(ty);
    opts.create_if_missing(true);
    let db = DB::open(&opts, &path);

    let should_open = match ty {
        DBCompressionType::None => true,
        DBCompressionType::Snappy => cfg!(feature = "snappy"),
        DBCompressionType::Zlib => cfg!(feature = "zlib"),
        DBCompressionType::Bz2 => cfg!(feature = "bzip2"),
        DBCompressionType::Lz4 | DBCompressionType::Lz4hc => cfg!(feature = "lz4"),
        DBCompressionType::Zstd => cfg!(feature = "zstd"),
    };

    if should_open {
        let _db = db.unwrap();
    } else {
        let _err = db.unwrap_err();
    }
}

#[test]
fn test_none_compression() {
    test_compression_type(DBCompressionType::None);
}

#[test]
fn test_snappy_compression() {
    test_compression_type(DBCompressionType::Snappy);
}

#[test]
fn test_zlib_compression() {
    test_compression_type(DBCompressionType::Zlib);
}

#[test]
fn test_bz2_compression() {
    test_compression_type(DBCompressionType::Bz2);
}

#[test]
fn test_lz4_compression() {
    test_compression_type(DBCompressionType::Lz4);
    test_compression_type(DBCompressionType::Lz4hc);
}

#[test]
fn test_zstd_compression() {
    test_compression_type(DBCompressionType::Zstd);
}

#[test]
fn test_add_compact_on_deletion_collector_factory() {
    let n = DBPath::new("_rust_rocksdb_test_add_compact_on_deletion_collector_factory");

    let mut opts = Options::default();
    opts.create_if_missing(true);
    opts.add_compact_on_deletion_collector_factory(5, 10, 0.5);
    let _db = DB::open(&opts, &n).unwrap();

    let mut rocksdb_log = fs::File::open(format!("{}/LOG", (&n).as_ref().to_str().unwrap()))
        .expect("rocksdb creates a LOG file");
    let mut settings = String::new();
    rocksdb_log
        .read_to_string(&mut settings)
        .expect("can read the LOG file");
    assert!(settings.contains("CompactOnDeletionCollector (Sliding window size = 5 Deletion trigger = 10 Deletion ratio = 0.5)"));
}

#[test]
fn test_set_avoid_unnecessary_blocking_io() {
    let path = DBPath::new("_set_avoid_unnecessary_blocking_io");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_avoid_unnecessary_blocking_io(true);
        let db = DB::open(&opts, &path).unwrap();
        let _ = db.put(b"k1", b"a");
        assert_eq!(&*db.get(b"k1").unwrap().unwrap(), b"a");
    }
}

#[test]
fn test_set_track_and_verify_wals_in_manifest() {
    let path = DBPath::new("_set_track_and_verify_wals_in_manifest");

    // test the defaults and the setter/accessor
    let mut opts = Options::default();
    assert!(!opts.get_track_and_verify_wals_in_manifest());
    opts.set_track_and_verify_wals_in_manifest(true);
    assert!(opts.get_track_and_verify_wals_in_manifest());
    opts.set_track_and_verify_wals_in_manifest(false);
    assert!(!opts.get_track_and_verify_wals_in_manifest());

    // verify that a database created with this option works
    // TODO: Check that the MANIFEST actually contains WalAddition/WalDeletion records
    opts.create_if_missing(true);
    opts.set_track_and_verify_wals_in_manifest(true);
    let db = DB::open(&opts, &path).unwrap();
    db.put(b"k1", b"a").expect("put must work");
    assert_eq!(db.get(b"k1").unwrap().unwrap(), b"a");
}

#[test]
fn test_set_periodic_compaction_seconds() {
    let path = DBPath::new("_set_periodic_compaction_seconds");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_periodic_compaction_seconds(5);
        let _db = DB::open(&opts, &path).unwrap();
    }
}

#[test]
fn test_set_ttl() {
    let path = DBPath::new("_set_ttl");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_ttl(5);
        let _db = DB::open(&opts, &path).unwrap();
    }
}

#[test]
fn test_set_ratelimiter() {
    let path = DBPath::new("_set_ratelimiter");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_ratelimiter(1024000, 1000, 1);
        let db = DB::open(&opts, &path).unwrap();

        let _ = db.put(b"k1", b"a");
        assert_eq!(&*db.get(b"k1").unwrap().unwrap(), b"a");
    }

    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_auto_tuned_ratelimiter(1024000, 1000, 1);
        let db = DB::open(&opts, &path).unwrap();

        let _ = db.put(b"k2", b"a");
        assert_eq!(&*db.get(b"k2").unwrap().unwrap(), b"a");
    }
}

#[test]
fn test_set_compaction_pri() {
    let path = DBPath::new("_set_compaction_pri");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_compaction_pri(DBCompactionPri::RoundRobin);
        let _db = DB::open(&opts, &path).unwrap();
    }
}

#[test]
fn test_set_blob_cache() {
    let path = DBPath::new("_set_blob_cache");
    let cache = Cache::new_hyper_clock_cache(1024 * 1024, 4 * 1024);

    let mut opts = Options::default();
    opts.create_if_missing(true);
    opts.set_enable_blob_files(true);
    opts.set_min_blob_size(16);
    opts.set_blob_cache(&cache);

    let db = DB::open(&opts, &path).unwrap();

    const KEY: &[u8] = b"k1";
    const VALUE: &[u8] = b"01234567890123456789";
    db.put(KEY, VALUE).unwrap();

    // Cache miss
    assert_eq!(&*db.get(KEY).unwrap().unwrap(), VALUE);

    // Cache hit
    assert_eq!(&*db.get(KEY).unwrap().unwrap(), VALUE);
}

#[test]
fn test_lru_cache_custom_opts() {
    let path = DBPath::new("_set_blob_cache");

    let mut lru_opts = LruCacheOptions::default();
    lru_opts.set_capacity(16 * 1024 * 1024);
    lru_opts.set_num_shard_bits(2);
    let cache = Cache::new_lru_cache_opts(&lru_opts);

    let mut opts = Options::default();
    opts.create_if_missing(true);
    opts.set_row_cache(&cache);

    // Must work even if we dropped the options: test that.
    drop(lru_opts);

    let db = DB::open(&opts, &path).unwrap();

    const KEY: &[u8] = b"k1";
    const VALUE: &[u8] = b"01234567890123456789";
    db.put(KEY, VALUE).unwrap();

    // Cache miss
    assert_eq!(&*db.get(KEY).unwrap().unwrap(), VALUE);

    // Cache hit
    assert_eq!(&*db.get(KEY).unwrap().unwrap(), VALUE);
}

struct UnsafeLoggerCheck {
    dropped: bool,
}

static UNSAFE_LOGGER_CHECK_DROP_COUNT: LazyLock<AtomicUsize> =
    LazyLock::new(|| AtomicUsize::new(0));

impl UnsafeLoggerCheck {
    fn new() -> Self {
        Self { dropped: false }
    }

    fn check_not_dropped(&self) {
        assert!(!self.dropped);
    }

    fn get_drop_count() -> usize {
        UNSAFE_LOGGER_CHECK_DROP_COUNT.load(std::sync::atomic::Ordering::SeqCst)
    }

    fn add_drop() {
        UNSAFE_LOGGER_CHECK_DROP_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
    }
}

impl Drop for UnsafeLoggerCheck {
    fn drop(&mut self) {
        self.dropped = true;
        Self::add_drop();
    }
}

fn static_callback_fn(level: rust_rocksdb::LogLevel, msg: &str) {
    println!("level={level:?} msg={msg}");
}

#[test]
fn test_set_callback_logger() {
    use rust_rocksdb::LogLevel::Debug;
    let path = DBPath::new("_set_callback_logger");

    let msgs = Arc::new(AtomicUsize::new(0));
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);

        // passing a static function compiles without any magic
        opts.set_callback_logger(rust_rocksdb::LogLevel::Error, static_callback_fn);

        // passing a closure should also work
        let closure_msgs = msgs.clone();
        let debug_msgs = Arc::new(AtomicUsize::new(0));
        let closure_debug_msgs = debug_msgs.clone();
        opts.set_callback_logger(Debug, move |level, _msg| {
            closure_msgs.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            if level == Debug {
                closure_debug_msgs.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            }
        });

        let _db = DB::open(&opts, &path).unwrap();
        // we expect some debug messages
        assert!(debug_msgs.load(std::sync::atomic::Ordering::SeqCst) > 0);
    }
    assert!(
        msgs.load(std::sync::atomic::Ordering::SeqCst) > 0,
        "callback logger produced no messages!"
    );

    // check that level filtering works: set the level to Info and expect 0 debug messages
    {
        let mut opts = Options::default();
        let debug_msgs = Arc::new(AtomicUsize::new(0));
        let closure_debug_msgs = debug_msgs.clone();
        opts.set_callback_logger(rust_rocksdb::LogLevel::Info, move |level, _msg| {
            if level == Debug {
                closure_debug_msgs.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            }
        });
        let _db = DB::open(&opts, &path).unwrap();

        // no debug messages: we set the level to Info: it should be excluded
        assert_eq!(debug_msgs.load(std::sync::atomic::Ordering::SeqCst), 0);
    }

    // test that the callback is not dropped prematurely and that cloning options still works
    msgs.store(0, std::sync::atomic::Ordering::SeqCst);
    assert_eq!(0, UnsafeLoggerCheck::get_drop_count());
    let opts = {
        let unsafe_logger_check = UnsafeLoggerCheck::new();
        let mut opts = Options::default();
        opts.create_if_missing(true);

        let closure_msgs = msgs.clone();
        opts.set_callback_logger(Debug, move |_lev, _msg| {
            unsafe_logger_check.check_not_dropped();
            closure_msgs.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        });
        opts
    };
    // UnsafeLoggerCheck is not dropped: it should have been moved
    assert_eq!(0, UnsafeLoggerCheck::get_drop_count());

    let opts_clone = opts.clone();

    assert_eq!(0, msgs.load(std::sync::atomic::Ordering::SeqCst));
    let _db = DB::open(&opts, &path).unwrap();
    drop(_db);
    assert!(
        msgs.load(std::sync::atomic::Ordering::SeqCst) > 0,
        "callback logger produced no messages!"
    );

    msgs.store(0, std::sync::atomic::Ordering::SeqCst);
    let _db2 = DB::open(&opts_clone, &path).unwrap();
    assert!(
        msgs.load(std::sync::atomic::Ordering::SeqCst) > 0,
        "callback logger produced no messages!"
    );
    drop(_db2);

    drop(opts);
    assert_eq!(0, UnsafeLoggerCheck::get_drop_count());
    drop(opts_clone);
    // finally UnsafeLoggerCheck must be dropped after all the options and the dbs are dropped
    assert_eq!(1, UnsafeLoggerCheck::get_drop_count());
}

#[test]
fn test_set_write_dbid_to_manifest() {
    let path = DBPath::new("_set_write_dbid_to_manifest");

    // test the defaults and the setter/accessor
    let mut opts = Options::default();
    assert!(opts.get_write_dbid_to_manifest());
    opts.set_write_dbid_to_manifest(false);
    assert!(!opts.get_write_dbid_to_manifest());
    opts.set_write_dbid_to_manifest(true);
    assert!(opts.get_write_dbid_to_manifest());

    // verify the DBID is preserved across checkpoints. If set to false this is not true
    opts.create_if_missing(true);
    opts.set_write_dbid_to_manifest(true);
    let db_orig = DB::open(&opts, &path).unwrap();
    let db_orig_id = db_orig.get_db_identity().unwrap();

    // a checkpoint from this database has the SAME DBID if it is in the manifest
    let checkpoint_path = DBPath::new("set_write_dbid_checkpoint");
    let checkpoint = Checkpoint::new(&db_orig).unwrap();
    checkpoint.create_checkpoint(&checkpoint_path).unwrap();

    let db_checkpoint = DB::open(&opts, &checkpoint_path).unwrap();
    let db_checkpoint_id = db_checkpoint.get_db_identity().unwrap();
    assert_eq!(
        db_orig_id,
        db_checkpoint_id,
        "expected database identity to be preserved across checkpoints; db_orig={} db_checkpoint={}",
        String::from_utf8_lossy(&db_orig_id),
        String::from_utf8_lossy(&db_checkpoint_id)
    );
}

#[test]
fn test_set_memtable_op_scan_flush_trigger() {
    let path = DBPath::new("_set_memtable_op_scan_flush_trigger");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_memtable_op_scan_flush_trigger(5);
        let _db = DB::open(&opts, &path).unwrap();
    }
}

#[test]
fn test_set_memtable_avg_op_scan_flush_trigger() {
    let path = DBPath::new("_set_memtable_avg_op_scan_flush_trigger");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_memtable_avg_op_scan_flush_trigger(5);
        let _db = DB::open(&opts, &path).unwrap();
    }
}

#[test]
fn jemalloc_init() {
    let path = DBPath::new("_jemalloc_init");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        let _db = DB::open(&opts, &path).unwrap();
    }

    let mut rocksdb_log =
        fs::File::open((&path).as_ref().join("LOG")).expect("rocksdb creates a LOG file");
    let mut log_content = String::new();
    rocksdb_log
        .read_to_string(&mut log_content)
        .expect("can read the LOG file");

    if cfg!(feature = "jemalloc")
        && !(
            // See NO_JEMALLOC_TARGETS in librocksdb-sys/build.rs
            cfg!(target_os = "android")
                || cfg!(target_os = "dragonfly")
                || cfg!(target_env = "musl")
                || cfg!(target_os = "macos")
                || cfg!(target_os = "ios")
        )
    {
        assert!(log_content.contains("Jemalloc supported: 1"));
    } else {
        assert!(log_content.contains("Jemalloc supported: 0"));
    }
}

// Verify we passed through the compiler flags to enable hardware CRC32 (if supported).
#[test]
fn test_crc32_build() {
    let path = DBPath::new("_crc32");
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        let _db = DB::open(&opts, &path).unwrap();
    }

    let mut rocksdb_log =
        fs::File::open((&path).as_ref().join("LOG")).expect("rocksdb creates a LOG file");
    let mut log_content = String::new();
    rocksdb_log
        .read_to_string(&mut log_content)
        .expect("can read the LOG file");

    // look for the log line with prefix "Fast CRC32 supported:"
    let log_line = log_content
        .lines()
        .find(|line| line.contains("Fast CRC32 supported:"));
    assert!(log_line.is_some(), "{log_content}");
    let log_line = log_line.unwrap();

    let expected_supported = if cfg!(target_arch = "x86_64") {
        // Default Linux x86-64 (x86-64-v1) does not support CRC32. Nearly all CPUs since ~2014
        // should support it (>= x86-64-v2), but RocksDB only does compile-time detection.
        // Our build.rs should match the Rust target settings
        if cfg!(target_feature = "crc") {
            Some(true)
        } else {
            Some(false)
        }
    } else if cfg!(target_arch = "aarch64") {
        // Default Linux aarch64 does not support CRC32. RocksDB's runtime feature detection is
        // also buggy as of 10.110 (2026-01-23), but it should report "Supported" if the feature is
        // enabled at compile time.
        if cfg!(target_feature = "crc") {
            Some(true)
        } else {
            // TODO: Fix when upstream RocksDB fixes the runtime feature detection
            Some(false)
        }
    } else {
        println!(
            "TODO: test_crc32_build needs to be extended to support ARCH={}",
            std::env::consts::ARCH
        );
        None
    };
    if let Some(expected_supported) = expected_supported {
        if expected_supported {
            assert!(
                log_line.contains("Supported on "),
                "expected 'Supported on ' log_line={log_line}"
            );
        } else {
            assert!(
                log_line.contains("Not supported on "),
                "expected 'Not supported on ' log_line={log_line}"
            );
        }
    }

    // split the last word as the supported arch
    let crc32_supported_arch = log_line.split_whitespace().last().unwrap();

    // Verify that the RocksDB reported architecture matches this test
    // this may fail if RocksDB changes its string formatting
    let expected_arch = if cfg!(target_arch = "x86_64") {
        "x86".to_string()
    } else if cfg!(target_arch = "aarch64") {
        // TODO: RocksDB has a bug: it can report x86 when the CRC feature is not enabled
        // This should just be "Arm64" when the RocksDB bug is fixed
        if cfg!(target_feature = "crc") {
            "Arm64".to_string()
        } else {
            "x86".to_string()
        }
    } else if cfg!(target_arch = "powerpc64") {
        "PPC".to_string()
    } else {
        format!("unknown Rust ARCH={} (fix test)", std::env::consts::ARCH)
    };

    assert_eq!(
        crc32_supported_arch, expected_arch,
        "Expected CRC32 support for architecture '{expected_arch}', but RocksDB reported support for '{crc32_supported_arch}'"
    );
}