diff --git a/src/binding/database/db_descriptor.cpp b/src/binding/database/db_descriptor.cpp index 7ed23d57..a770c15c 100644 --- a/src/binding/database/db_descriptor.cpp +++ b/src/binding/database/db_descriptor.cpp @@ -55,6 +55,27 @@ static void applyCompression( level ? *level : rocksdb::CompressionOptions::kDefaultCompressionLevel; } +rocksdb::ColumnFamilyOptions buildColumnFamilyOptions( + const DBOptions& options, + rocksdb::ColumnFamilyOptions cfOptions +) { + rocksdb::BlockBasedTableOptions tableOptions; + if (options.noBlockCache) { + tableOptions.no_block_cache = true; + } else { + tableOptions.block_cache = DBSettings::getInstance().getBlockCache(); + } + + cfOptions.enable_blob_files = true; + cfOptions.min_blob_size = 2048; + cfOptions.enable_blob_garbage_collection = true; + cfOptions.write_buffer_size = static_cast(options.writeBufferSize); + cfOptions.max_write_buffer_number = options.maxWriteBufferNumber; + cfOptions.max_write_buffer_size_to_maintain = options.maxWriteBufferSizeToMaintain; + cfOptions.table_factory.reset(rocksdb::NewBlockBasedTableFactory(tableOptions)); + return cfOptions; +} + // Reads each existing column family's persisted compression from the database's // latest OPTIONS file into `result`, returning the RocksDB status. The OPTIONS // file is the ONLY authoritative source for a CF's stored compression (RocksDB @@ -190,6 +211,7 @@ class TransactionLogEventListener : public rocksdb::EventListener { DBDescriptor::DBDescriptor( const std::string& path, const DBOptions& options, + const rocksdb::ColumnFamilyOptions& cfOptions, std::shared_ptr db, std::unordered_map>&& columns, std::shared_ptr statistics @@ -198,6 +220,7 @@ DBDescriptor::DBDescriptor( vtEpoch(nextVtEpoch()), mode(options.mode), readOnly(options.readOnly), + cfOptions(cfOptions), db(db), columns(std::move(columns)), statistics(statistics) @@ -854,14 +877,7 @@ std::shared_ptr DBDescriptor::open(const std::string& path, const std::string name = options.name.empty() ? "default" : options.name; DEBUG_LOG("DBDescriptor::open Opening \"%s\" (column family: \"%s\", read-only: %s)\n", path.c_str(), name.c_str(), options.readOnly ? "true" : "false"); - // set or disable the block cache DBSettings& settings = DBSettings::getInstance(); - rocksdb::BlockBasedTableOptions tableOptions; - if (options.noBlockCache) { - tableOptions.no_block_cache = true; - } else { - tableOptions.block_cache = settings.getBlockCache(); - } // set the database options rocksdb::Options dbOptions; @@ -894,17 +910,9 @@ std::shared_ptr DBDescriptor::open(const std::string& path, const dbOptions.statistics = nullptr; } - // Base ColumnFamilyOptions (blob + write-buffer + table settings) shared by - // every column family. Compression is intentionally NOT set here: it is - // per-CF, so it is applied per descriptor below. - rocksdb::ColumnFamilyOptions cfOptions; - cfOptions.enable_blob_files = true; - cfOptions.min_blob_size = 2048; - cfOptions.enable_blob_garbage_collection = true; - cfOptions.write_buffer_size = static_cast(options.writeBufferSize); - cfOptions.max_write_buffer_number = options.maxWriteBufferNumber; - cfOptions.max_write_buffer_size_to_maintain = options.maxWriteBufferSizeToMaintain; - cfOptions.table_factory.reset(rocksdb::NewBlockBasedTableFactory(tableOptions)); + // Base options shared by every column family. Compression is applied per CF + // below so opening one family cannot restamp another's algorithm. + auto cfOptions = buildColumnFamilyOptions(options); // create a shared pointer to hold the weak descriptor reference for the event listener auto descriptorWeakPtr = std::make_shared>(); @@ -1019,13 +1027,17 @@ std::shared_ptr DBDescriptor::open(const std::string& path, const } } if (!columnExists) { - auto column = rocksdb_js::createRocksDBColumnFamily(db, options.name, options.compression, options.compressionLevel); + auto cfo = cfOptions; + if (options.compression) { + applyCompression(cfo, *options.compression, options.compressionLevel); + } + auto column = rocksdb_js::createRocksDBColumnFamily(db, options.name, cfo); auto columnDescriptor = std::make_shared(column); columns[options.name] = columnDescriptor; } DEBUG_LOG("DBDescriptor::open Creating DBDescriptor for \"%s\"\n", path.c_str()); - auto descriptor = std::shared_ptr(new DBDescriptor(path, options, db, std::move(columns), dbOptions.statistics)); + auto descriptor = std::shared_ptr(new DBDescriptor(path, options, cfOptions, db, std::move(columns), dbOptions.statistics)); // set the weak pointer for the event listener *descriptorWeakPtr = descriptor; diff --git a/src/binding/database/db_descriptor.h b/src/binding/database/db_descriptor.h index 3d5d1ec8..d733435c 100644 --- a/src/binding/database/db_descriptor.h +++ b/src/binding/database/db_descriptor.h @@ -33,6 +33,11 @@ struct TransactionHandle; struct UserSharedBufferData; struct UserSharedBufferFinalizeData; +rocksdb::ColumnFamilyOptions buildColumnFamilyOptions( + const DBOptions& options, + rocksdb::ColumnFamilyOptions cfOptions = {} +); + /** * Custom deleter for RocksDB that waits for any background compaction to * complete before destroying the database instance. Compaction is triggered @@ -92,6 +97,13 @@ struct DBDescriptor final : public std::enable_shared_from_this { */ bool readOnly; + /** + * Base column family options retained from `DB::Open`. Families created + * later preserve these table/blob settings while applying the current + * handle's per-CF memory options. + */ + rocksdb::ColumnFamilyOptions cfOptions; + /** * The RocksDB database instance. */ @@ -259,6 +271,7 @@ struct DBDescriptor final : public std::enable_shared_from_this { DBDescriptor( const std::string& path, const DBOptions& options, + const rocksdb::ColumnFamilyOptions& cfOptions, std::shared_ptr db, std::unordered_map>&& columns, std::shared_ptr statistics diff --git a/src/binding/database/db_registry.cpp b/src/binding/database/db_registry.cpp index 9709b63a..792dd547 100644 --- a/src/binding/database/db_registry.cpp +++ b/src/binding/database/db_registry.cpp @@ -334,7 +334,19 @@ std::unique_ptr DBRegistry::OpenDB(const std::string& path, cons throw rocksdb_js::DBException("Column family \"" + name + "\" not found: cannot create column family in read-only mode"); } DEBUG_LOG("%p DBRegistry::OpenDB Creating column family \"%s\"\n", instance.get(), name.c_str()); - auto column = rocksdb_js::createRocksDBColumnFamily(entry.descriptor->db, name, options.compression, options.compressionLevel); + // Preserve retained settings while applying every per-CF option from + // the handle creating this family. + auto cfOptions = buildColumnFamilyOptions(options, entry.descriptor->cfOptions); + if (options.compression) { + cfOptions.compression = *options.compression; + cfOptions.blob_compression_type = *options.compression; + cfOptions.compression_opts.level = options.compressionLevel + ? *options.compressionLevel + : rocksdb::CompressionOptions::kDefaultCompressionLevel; + } + auto column = rocksdb_js::createRocksDBColumnFamily( + entry.descriptor->db, name, cfOptions + ); auto columnDescriptor = std::make_shared(column); columns[name] = columnDescriptor; entry.descriptor->columns[name] = columnDescriptor; diff --git a/src/binding/napi/helpers.cpp b/src/binding/napi/helpers.cpp index 4144d947..c81a68e4 100644 --- a/src/binding/napi/helpers.cpp +++ b/src/binding/napi/helpers.cpp @@ -14,7 +14,6 @@ #include "napi/helpers.h" #include "napi/macros.h" #include "rocksdb/utilities/options_util.h" -#include "database/db_settings.h" namespace rocksdb_js { @@ -232,30 +231,9 @@ std::string getNapiExtendedError(napi_env env, napi_status& status, const char* std::shared_ptr createRocksDBColumnFamily( const std::shared_ptr db, const std::string& name, - const std::optional& compression, - const std::optional& compressionLevel + const rocksdb::ColumnFamilyOptions& cfOptions ) { rocksdb::ColumnFamilyHandle* cfHandle; - rocksdb::BlockBasedTableOptions tableOptions; - DBSettings& settings = DBSettings::getInstance(); - tableOptions.block_cache = settings.getBlockCache(); - rocksdb::ColumnFamilyOptions cfOptions; - cfOptions.enable_blob_files = true; - cfOptions.min_blob_size = 2048; - cfOptions.enable_blob_garbage_collection = true; - if (compression) { - cfOptions.compression = *compression; - // Large values are stored in blob files (enable_blob_files), which have - // their own compression setting that defaults to none. Apply the same - // algorithm there so the option compresses the whole dataset, not just the - // inline (< min_blob_size) portion. - cfOptions.blob_compression_type = *compression; - if (compressionLevel) { - cfOptions.compression_opts.level = *compressionLevel; - } - } - cfOptions.table_factory.reset(rocksdb::NewBlockBasedTableFactory(tableOptions)); - rocksdb::Status status = db->CreateColumnFamily(cfOptions, name, &cfHandle); if (!status.ok()) { throw rocksdb_js::DBException(status.ToString()); diff --git a/src/binding/napi/helpers.h b/src/binding/napi/helpers.h index 58182857..5d0ff568 100644 --- a/src/binding/napi/helpers.h +++ b/src/binding/napi/helpers.h @@ -25,11 +25,15 @@ namespace rocksdb_js { void createJSError(napi_env env, const char* code, const char* message, napi_value& error); +/** + * Creates a column family on an already-open database. Takes the database's own + * `ColumnFamilyOptions`, which only reach families that existed at `DB::Open`; + * rebuilding a set here would silently drop every configured tunable. + */ std::shared_ptr createRocksDBColumnFamily( const std::shared_ptr db, const std::string& name, - const std::optional& compression = std::nullopt, - const std::optional& compressionLevel = std::nullopt + const rocksdb::ColumnFamilyOptions& cfOptions ); void createRocksDBError(napi_env env, rocksdb::Status status, const char* msg, napi_value& error); diff --git a/test/block-cache.test.ts b/test/block-cache.test.ts index 9d2979fe..94211b4c 100644 --- a/test/block-cache.test.ts +++ b/test/block-cache.test.ts @@ -28,6 +28,17 @@ describe('Block Cache', () => { expect(db.get('foo')).toBe('bar'); })); + it('should apply noBlockCache to a late-created column family', () => + dbRunner({ dbOptions: [{}, { name: 'late', noBlockCache: true }] }, async (_, { db }) => { + await db.put('foo', 'bar'); + await db.flush(); + + const firstRead = db.get('foo'); + expect(firstRead).toBeInstanceOf(Promise); + expect(await firstRead).toBe('bar'); + expect(db.get('foo')).toBeInstanceOf(Promise); + })); + it('should change the block cache size', () => dbRunner({ dbOptions: [{ noBlockCache: true }], skipOpen: true }, async ({ db }) => { RocksDatabase.config({ blockCacheSize: 1024 * 1024 }); diff --git a/test/db-options.test.ts b/test/db-options.test.ts index d4d5d5ad..83ad9ddd 100644 --- a/test/db-options.test.ts +++ b/test/db-options.test.ts @@ -1,4 +1,4 @@ -import { dbRunner } from './lib/util.js'; +import { dbRunner, generateDBPath } from './lib/util.js'; import { describe, expect, it } from 'vitest'; describe('Database write buffer options', () => { @@ -75,6 +75,71 @@ describe('Database write buffer options', () => { expect(() => db.open()).toThrow('maxOpenFiles must be'); })); + // A named CF is created after DB::Open, so it misses the open options. + // Observed via flush behavior: RocksDB exposes no property for the size. + it('should apply writeBufferSize to a named column family', () => + dbRunner( + { + dbOptions: [ + { + path: generateDBPath(), + name: 'mycf', + writeBufferSize: 64 * 1024, + maxWriteBufferNumber: 2, + }, + { + path: generateDBPath(), + name: 'mycf', + writeBufferSize: 64 * 1024 * 1024, + maxWriteBufferNumber: 2, + }, + ], + }, + async ({ db: smallBuffer }, { db: largeBuffer }) => { + const value = 'x'.repeat(1024); + for (const db of [smallBuffer, largeBuffer]) { + for (let i = 0; i < 512; i++) { + await db.put(`key-${i.toString().padStart(6, '0')}`, value); + } + } + + // With only two 64 KiB memtables, all writes cannot complete until + // RocksDB flushes at least one; write backpressure is the gate. + const smallBufferSize = smallBuffer.getDBIntProperty('rocksdb.total-sst-files-size') ?? 0; + expect(smallBufferSize).toBeGreaterThan(0); + expect(largeBuffer.getDBIntProperty('rocksdb.total-sst-files-size')).toBe(0); + } + )); + + it('should apply the current handle memory options to a late column family', () => + dbRunner( + { + dbOptions: [ + { writeBufferSize: 64 * 1024 * 1024, maxWriteBufferNumber: 2 }, + { + name: 'late', + writeBufferSize: 64 * 1024, + maxWriteBufferNumber: 2, + }, + ], + skipOpen: true, + }, + async ({ db: first }, { db: late }) => { + first.open(); + late.open(); + + const value = 'x'.repeat(1024); + for (let i = 0; i < 512; i++) { + await late.put(`key-${i.toString().padStart(6, '0')}`, value); + } + + // The two-buffer limit gates on automatic flush without forcing one; + // an incorrectly inherited 64 MiB buffer would still have no SST. + const lateBufferSize = late.getDBIntProperty('rocksdb.total-sst-files-size') ?? 0; + expect(lateBufferSize).toBeGreaterThan(0); + } + )); + it('should flush memtables when writeBufferSize is exceeded', () => dbRunner({ dbOptions: [{ writeBufferSize: 64 * 1024 }] }, async ({ db }) => { // 64KB memtable; write enough data to force at least one flush.