diff --git a/build.sh b/build.sh
index 4e6472a..0dc2f26 100644
--- a/build.sh
+++ b/build.sh
@@ -33,7 +33,7 @@ cd $BUILD_PATH && wget https://github.com/facebook/zstd/archive/v${zstd_version}
     -DCMAKE_POSITION_INDEPENDENT_CODE=ON -DZSTD_ZLIB_SUPPORT=ON -DZSTD_LZMA_SUPPORT=OFF -DCMAKE_BUILD_TYPE=Release .. && make -j$(nproc) install && \
     cd $BUILD_PATH && rm -rf * && ldconfig
 
-rocksdb_version="6.22.1"
+rocksdb_version="6.23.3"
 cd $BUILD_PATH && wget https://github.com/facebook/rocksdb/archive/v${rocksdb_version}.tar.gz && tar xzf v${rocksdb_version}.tar.gz && cd rocksdb-${rocksdb_version}/ && \
     mkdir -p build_place && cd build_place && cmake -DCMAKE_BUILD_TYPE=Release $CMAKE_REQUIRED_PARAMS -DCMAKE_PREFIX_PATH=$INSTALL_PREFIX -DWITH_TESTS=OFF -DWITH_GFLAGS=OFF \
     -DWITH_BENCHMARK_TOOLS=OFF -DWITH_TOOLS=OFF -DWITH_MD_LIBRARY=OFF -DWITH_RUNTIME_DEBUG=OFF -DROCKSDB_BUILD_SHARED=OFF -DWITH_SNAPPY=ON -DWITH_LZ4=ON -DWITH_ZLIB=ON \
diff --git a/dist/linux_amd64/include/rocksdb/advanced_options.h b/dist/linux_amd64/include/rocksdb/advanced_options.h
index 7804ec4..1c4f09b 100644
--- a/dist/linux_amd64/include/rocksdb/advanced_options.h
+++ b/dist/linux_amd64/include/rocksdb/advanced_options.h
@@ -173,9 +173,10 @@ struct CompressionOptions {
         parallel_threads(1),
         enabled(false),
         max_dict_buffer_bytes(0) {}
-  CompressionOptions(int wbits, int _lev, int _strategy, int _max_dict_bytes,
-                     int _zstd_max_train_bytes, int _parallel_threads,
-                     bool _enabled, uint64_t _max_dict_buffer_bytes)
+  CompressionOptions(int wbits, int _lev, int _strategy,
+                     uint32_t _max_dict_bytes, uint32_t _zstd_max_train_bytes,
+                     uint32_t _parallel_threads, bool _enabled,
+                     uint64_t _max_dict_buffer_bytes)
       : window_bits(wbits),
         level(_lev),
         strategy(_strategy),
diff --git a/dist/linux_amd64/include/rocksdb/c.h b/dist/linux_amd64/include/rocksdb/c.h
index 8280b25..51fe928 100644
--- a/dist/linux_amd64/include/rocksdb/c.h
+++ b/dist/linux_amd64/include/rocksdb/c.h
@@ -1602,6 +1602,8 @@ extern ROCKSDB_LIBRARY_API rocksdb_filterpolicy_t*
 rocksdb_filterpolicy_create_bloom(int bits_per_key);
 extern ROCKSDB_LIBRARY_API rocksdb_filterpolicy_t*
 rocksdb_filterpolicy_create_bloom_full(int bits_per_key);
+extern ROCKSDB_LIBRARY_API rocksdb_filterpolicy_t*
+rocksdb_filterpolicy_create_ribbon(int bloom_equivalent_bits_per_key);
 
 /* Merge Operator */
 
diff --git a/dist/linux_amd64/include/rocksdb/comparator.h b/dist/linux_amd64/include/rocksdb/comparator.h
index 37c2925..fe96eb0 100644
--- a/dist/linux_amd64/include/rocksdb/comparator.h
+++ b/dist/linux_amd64/include/rocksdb/comparator.h
@@ -10,6 +10,7 @@
 
 #include <string>
 
+#include "rocksdb/customizable.h"
 #include "rocksdb/rocksdb_namespace.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -20,7 +21,7 @@ class Slice;
 // used as keys in an sstable or a database.  A Comparator implementation
 // must be thread-safe since rocksdb may invoke its methods concurrently
 // from multiple threads.
-class Comparator {
+class Comparator : public Customizable {
  public:
   Comparator() : timestamp_size_(0) {}
 
@@ -37,7 +38,11 @@ class Comparator {
 
   virtual ~Comparator() {}
 
+  static Status CreateFromString(const ConfigOptions& opts,
+                                 const std::string& id,
+                                 const Comparator** comp);
   static const char* Type() { return "Comparator"; }
+
   // Three-way comparison.  Returns value:
   //   < 0 iff "a" < "b",
   //   == 0 iff "a" == "b",
diff --git a/dist/linux_amd64/include/rocksdb/configurable.h b/dist/linux_amd64/include/rocksdb/configurable.h
index b56072d..8c90dd7 100644
--- a/dist/linux_amd64/include/rocksdb/configurable.h
+++ b/dist/linux_amd64/include/rocksdb/configurable.h
@@ -8,6 +8,7 @@
 
 #pragma once
 
+#include <atomic>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
@@ -55,7 +56,7 @@ class Configurable {
   };
 
  public:
-  Configurable() : prepared_(false) {}
+  Configurable();
   virtual ~Configurable() {}
 
   // Returns the raw pointer of the named options that is used by this
@@ -264,12 +265,12 @@ class Configurable {
   // Returns true if this object has been initialized via PrepareOptions, false
   // otherwise. Once an object has been prepared, only mutable options may be
   // changed.
-  virtual bool IsPrepared() const { return prepared_; }
+  virtual bool IsPrepared() const { return is_prepared_; }
 
  protected:
   // True once the object is prepared.  Once the object is prepared, only
   // mutable options can be configured.
-  bool prepared_;
+  std::atomic<bool> is_prepared_;
 
   // Returns the raw pointer for the associated named option.
   // The name is typically the name of an option registered via the
diff --git a/dist/linux_amd64/include/rocksdb/customizable.h b/dist/linux_amd64/include/rocksdb/customizable.h
index 24ddfa5..c04e713 100644
--- a/dist/linux_amd64/include/rocksdb/customizable.h
+++ b/dist/linux_amd64/include/rocksdb/customizable.h
@@ -138,6 +138,40 @@ class Customizable : public Configurable {
   Status GetOption(const ConfigOptions& config_options, const std::string& name,
                    std::string* value) const override;
 #endif  // ROCKSDB_LITE
+  // Helper method for getting for parsing the opt_value into the corresponding
+  // options for use in potentially creating a new Customizable object (this
+  // method is primarily a support method for LoadSharedObject et al for new
+  // Customizable objects). The opt_value may be either name-value pairs
+  // separated by ";" (a=b; c=d), or a simple name (a). In order to create a new
+  // Customizable, the ID is determined by:
+  // - If the value is a simple name (e.g. "BlockBasedTable"), the id is this
+  // name;
+  // - Otherwise, if there is a "id=value", the id is set to "value"
+  // - Otherwise, if the input customizable is not null, custom->GetId is used
+  // - Otherwise, an error is returned.
+  //
+  // If the opt_value is name-value pairs, these pairs will be returned in
+  // options (without the id pair). If the ID being returned matches the ID of
+  // the input custom object, then the options from the input object will also
+  // be added to the returned options.
+  //
+  // This method returns non-OK if the ID could not be found, or if the
+  // opt_value could not be parsed into name-value pairs.
+  static Status GetOptionsMap(
+      const ConfigOptions& config_options, const Customizable* custom,
+      const std::string& opt_value, std::string* id,
+      std::unordered_map<std::string, std::string>* options);
+
+  // Helper method to configure a new object with the supplied options.
+  // If the object is not null and invoke_prepare_options=true, the object
+  // will be configured and prepared.
+  // Returns success if the object is properly configured and (optionally)
+  // prepared Returns InvalidArgument if the object is nullptr and there are
+  // options in the map Returns the result of the ConfigureFromMap or
+  // PrepareOptions
+  static Status ConfigureNewObject(
+      const ConfigOptions& config_options, Customizable* object,
+      const std::unordered_map<std::string, std::string>& options);
 
   // Returns the inner class when a Customizable implements a has-a (wrapped)
   // relationship.  Derived classes that implement a has-a must override this
diff --git a/dist/linux_amd64/include/rocksdb/db.h b/dist/linux_amd64/include/rocksdb/db.h
index a8c3bbd..645ecfb 100644
--- a/dist/linux_amd64/include/rocksdb/db.h
+++ b/dist/linux_amd64/include/rocksdb/db.h
@@ -1423,6 +1423,12 @@ class DB {
     GetColumnFamilyMetaData(DefaultColumnFamily(), metadata);
   }
 
+  // Obtains the meta data of all column families for the DB.
+  // The returned map contains one entry for each column family indexed by the
+  // name of the column family.
+  virtual void GetAllColumnFamilyMetaData(
+      std::vector<ColumnFamilyMetaData>* /*metadata*/) {}
+
   // IngestExternalFile() will load a list of external SST files (1) into the DB
   // Two primary modes are supported:
   // - Duplicate keys in the new files will overwrite exiting keys (default)
diff --git a/dist/linux_amd64/include/rocksdb/env_encryption.h b/dist/linux_amd64/include/rocksdb/env_encryption.h
index 7a76ec8..e243e3e 100644
--- a/dist/linux_amd64/include/rocksdb/env_encryption.h
+++ b/dist/linux_amd64/include/rocksdb/env_encryption.h
@@ -9,6 +9,7 @@
 
 #include <string>
 
+#include "rocksdb/customizable.h"
 #include "rocksdb/env.h"
 #include "rocksdb/file_system.h"
 #include "rocksdb/rocksdb_namespace.h"
@@ -58,7 +59,7 @@ class BlockAccessCipherStream {
 };
 
 // BlockCipher
-class BlockCipher {
+class BlockCipher : public Customizable {
  public:
   virtual ~BlockCipher(){};
 
@@ -80,12 +81,12 @@ class BlockCipher {
                                  const std::string& value,
                                  std::shared_ptr<BlockCipher>* result);
 
+  static const char* Type() { return "BlockCipher"; }
   // Short-cut method to create a ROT13 BlockCipher.
   // This cipher is only suitable for test purposes and should not be used in
   // production!!!
   static std::shared_ptr<BlockCipher> NewROT13Cipher(size_t block_size);
 
-  virtual const char* Name() const = 0;
   // BlockSize returns the size of each block supported by this cipher stream.
   virtual size_t BlockSize() = 0;
 
@@ -101,7 +102,7 @@ class BlockCipher {
 // The encryption provider is used to create a cipher stream for a specific
 // file. The returned cipher stream will be used for actual
 // encryption/decryption actions.
-class EncryptionProvider {
+class EncryptionProvider : public Customizable {
  public:
   virtual ~EncryptionProvider(){};
 
@@ -109,14 +110,14 @@ class EncryptionProvider {
   // The value describes the type of provider (and potentially optional
   // configuration parameters) used to create this provider.
   // For example, if the value is "CTR", a CTREncryptionProvider will be
-  // created. If the value is preceded by "test://" (e.g test://CTR"), the
-  // TEST_Initialize method will be invoked prior to returning the provider.
+  // created. If the value is ends with "://test" (e.g CTR://test"), the
+  // provider will be initialized in "TEST" mode prior to being returned.
   //
   // @param config_options  Options to control how this provider is created
   //                        and initialized.
   // @param value  The value might be:
   //   - CTR         Create a CTR provider
-  //   - test://CTR Create a CTR provider and initialize it for tests.
+  //   - CTR://test Create a CTR provider and initialize it for tests.
   // @param result The new provider object
   // @return OK if the provider was successfully created
   // @return NotFound if an invalid name was specified in the value
@@ -125,13 +126,12 @@ class EncryptionProvider {
                                  const std::string& value,
                                  std::shared_ptr<EncryptionProvider>* result);
 
+  static const char* Type() { return "EncryptionProvider"; }
+
   // Short-cut method to create a CTR-provider
   static std::shared_ptr<EncryptionProvider> NewCTRProvider(
       const std::shared_ptr<BlockCipher>& cipher);
 
-  // Returns the name of this EncryptionProvider
-  virtual const char* Name() const = 0;
-
   // GetPrefixLength returns the length of the prefix that is added to every
   // file and used for storing encryption options. For optimal performance, the
   // prefix length should be a multiple of the page size.
@@ -165,11 +165,6 @@ class EncryptionProvider {
   // or not a file is encrypted by this provider.  The maker will also be part
   // of any encryption prefix for this provider.
   virtual std::string GetMarker() const { return ""; }
-
- protected:
-  // Optional method to initialize an EncryptionProvider in the TEST
-  // environment.
-  virtual Status TEST_Initialize() { return Status::OK(); }
 };
 
 class EncryptedSequentialFile : public FSSequentialFile {
diff --git a/dist/linux_amd64/include/rocksdb/flush_block_policy.h b/dist/linux_amd64/include/rocksdb/flush_block_policy.h
index badc080..38fc633 100644
--- a/dist/linux_amd64/include/rocksdb/flush_block_policy.h
+++ b/dist/linux_amd64/include/rocksdb/flush_block_policy.h
@@ -6,12 +6,15 @@
 #pragma once
 
 #include <string>
+
+#include "rocksdb/customizable.h"
 #include "rocksdb/table.h"
 
 namespace ROCKSDB_NAMESPACE {
 
 class Slice;
 class BlockBuilder;
+struct ConfigOptions;
 struct Options;
 
 // FlushBlockPolicy provides a configurable way to determine when to flush a
@@ -25,10 +28,16 @@ class FlushBlockPolicy {
   virtual ~FlushBlockPolicy() {}
 };
 
-class FlushBlockPolicyFactory {
+class FlushBlockPolicyFactory : public Customizable {
  public:
-  // Return the name of the flush block policy.
-  virtual const char* Name() const = 0;
+  static const char* Type() { return "FlushBlockPolicyFactory"; }
+
+  // Creates a FlushBlockPolicyFactory based on the input value.
+  // By default, this method can create EveryKey or BySize PolicyFactory,
+  // which take now config_options.
+  static Status CreateFromString(
+      const ConfigOptions& config_options, const std::string& value,
+      std::shared_ptr<FlushBlockPolicyFactory>* result);
 
   // Return a new block flush policy that flushes data blocks by data size.
   // FlushBlockPolicy may need to access the metadata of the data block
@@ -45,9 +54,10 @@ class FlushBlockPolicyFactory {
 
 class FlushBlockBySizePolicyFactory : public FlushBlockPolicyFactory {
  public:
-  FlushBlockBySizePolicyFactory() {}
+  FlushBlockBySizePolicyFactory();
 
-  const char* Name() const override { return "FlushBlockBySizePolicyFactory"; }
+  static const char* kClassName() { return "FlushBlockBySizePolicyFactory"; }
+  const char* Name() const override { return kClassName(); }
 
   FlushBlockPolicy* NewFlushBlockPolicy(
       const BlockBasedTableOptions& table_options,
diff --git a/dist/linux_amd64/include/rocksdb/metadata.h b/dist/linux_amd64/include/rocksdb/metadata.h
index b515c51..25fcdde 100644
--- a/dist/linux_amd64/include/rocksdb/metadata.h
+++ b/dist/linux_amd64/include/rocksdb/metadata.h
@@ -15,6 +15,7 @@
 #include "rocksdb/types.h"
 
 namespace ROCKSDB_NAMESPACE {
+struct BlobMetaData;
 struct ColumnFamilyMetaData;
 struct LevelMetaData;
 struct SstFileMetaData;
@@ -35,6 +36,13 @@ struct ColumnFamilyMetaData {
   std::string name;
   // The metadata of all levels in this column family.
   std::vector<LevelMetaData> levels;
+
+  // The total size of all blob files
+  uint64_t blob_file_size = 0;
+  // The number of blob files in this column family.
+  size_t blob_file_count = 0;
+  // The metadata of the blobs in this column family
+  std::vector<BlobMetaData> blob_files;
 };
 
 // The metadata that describes a level.
@@ -153,6 +161,44 @@ struct LiveFileMetaData : SstFileMetaData {
   LiveFileMetaData() : column_family_name(), level(0) {}
 };
 
+// The MetaData that describes a Blob file
+struct BlobMetaData {
+  BlobMetaData()
+      : blob_file_number(0),
+        blob_file_size(0),
+        total_blob_count(0),
+        total_blob_bytes(0),
+        garbage_blob_count(0),
+        garbage_blob_bytes(0) {}
+
+  BlobMetaData(uint64_t _file_number, const std::string& _file_name,
+               const std::string& _file_path, uint64_t _file_size,
+               uint64_t _total_blob_count, uint64_t _total_blob_bytes,
+               uint64_t _garbage_blob_count, uint64_t _garbage_blob_bytes,
+               const std::string& _file_checksum,
+               const std::string& _file_checksum_func_name)
+      : blob_file_number(_file_number),
+        blob_file_name(_file_name),
+        blob_file_path(_file_path),
+        blob_file_size(_file_size),
+        total_blob_count(_total_blob_count),
+        total_blob_bytes(_total_blob_bytes),
+        garbage_blob_count(_garbage_blob_count),
+        garbage_blob_bytes(_garbage_blob_bytes),
+        checksum_method(_file_checksum),
+        checksum_value(_file_checksum_func_name) {}
+  uint64_t blob_file_number;
+  std::string blob_file_name;
+  std::string blob_file_path;
+  uint64_t blob_file_size;
+  uint64_t total_blob_count;
+  uint64_t total_blob_bytes;
+  uint64_t garbage_blob_count;
+  uint64_t garbage_blob_bytes;
+  std::string checksum_method;
+  std::string checksum_value;
+};
+
 // Metadata returned as output from ExportColumnFamily() and used as input to
 // CreateColumnFamiliesWithImport().
 struct ExportImportFilesMetaData {
diff --git a/dist/linux_amd64/include/rocksdb/options.h b/dist/linux_amd64/include/rocksdb/options.h
index 4b46a75..220f904 100644
--- a/dist/linux_amd64/include/rocksdb/options.h
+++ b/dist/linux_amd64/include/rocksdb/options.h
@@ -211,14 +211,18 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
   CompressionType compression;
 
   // Compression algorithm that will be used for the bottommost level that
-  // contain files.
+  // contain files. The behavior for num_levels = 1 is not well defined.
+  // Right now, with num_levels = 1,  all compaction outputs will use
+  // bottommost_compression and all flush outputs still use options.compression,
+  // but the behavior is subject to change.
   //
   // Default: kDisableCompressionOption (Disabled)
   CompressionType bottommost_compression = kDisableCompressionOption;
 
   // different options for compression algorithms used by bottommost_compression
   // if it is enabled. To enable it, please see the definition of
-  // CompressionOptions.
+  // CompressionOptions. Behavior for num_levels = 1 is the same as
+  // options.bottommost_compression.
   CompressionOptions bottommost_compression_opts;
 
   // different options for compression algorithms
@@ -384,12 +388,12 @@ class CompactionService : public Customizable {
   // TODO: sub-compaction is not supported, as they will have the same job_id, a
   // sub-compaction id might be added
   virtual CompactionServiceJobStatus Start(
-      const std::string& compaction_service_input, int job_id) = 0;
+      const std::string& compaction_service_input, uint64_t job_id) = 0;
 
   // Wait compaction to be finish.
   // TODO: Add output path override
   virtual CompactionServiceJobStatus WaitForComplete(
-      int job_id, std::string* compaction_service_result) = 0;
+      uint64_t job_id, std::string* compaction_service_result) = 0;
 
   virtual ~CompactionService() {}
 };
@@ -778,6 +782,10 @@ struct DBOptions {
   // Default: true
   bool advise_random_on_open = true;
 
+  // If true, allows for memtable purge instead of flush to storage.
+  // (experimental).
+  bool experimental_allow_mempurge = false;
+
   // Amount of data to build up in memtables across all column
   // families before writing to disk.
   //
@@ -1382,7 +1390,7 @@ struct ReadOptions {
   // Default: true
   bool verify_checksums;
 
-  // Should the "data block"/"index block"" read for this iteration be placed in
+  // Should the "data block"/"index block" read for this iteration be placed in
   // block cache?
   // Callers may wish to set this field to false for bulk scans.
   // This would help not to the change eviction order of existing items in the
@@ -1797,6 +1805,11 @@ struct CompactionServiceOptionsOverride {
   std::shared_ptr<const SliceTransform> prefix_extractor = nullptr;
   std::shared_ptr<TableFactory> table_factory;
   std::shared_ptr<SstPartitionerFactory> sst_partitioner_factory = nullptr;
+
+  // statistics is used to collect DB operation metrics, the metrics won't be
+  // returned to CompactionService primary host, to collect that, the user needs
+  // to set it here.
+  std::shared_ptr<Statistics> statistics = nullptr;
 };
 
 }  // namespace ROCKSDB_NAMESPACE
diff --git a/dist/linux_amd64/include/rocksdb/secondary_cache.h b/dist/linux_amd64/include/rocksdb/secondary_cache.h
index 221b3e5..20632d2 100644
--- a/dist/linux_amd64/include/rocksdb/secondary_cache.h
+++ b/dist/linux_amd64/include/rocksdb/secondary_cache.h
@@ -10,6 +10,7 @@
 #include <string>
 
 #include "rocksdb/cache.h"
+#include "rocksdb/customizable.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/statistics.h"
 #include "rocksdb/status.h"
@@ -42,13 +43,14 @@ class SecondaryCacheResultHandle {
 //
 // Cache interface for caching blocks on a secondary tier (which can include
 // non-volatile media, or alternate forms of caching such as compressed data)
-class SecondaryCache {
+class SecondaryCache : public Customizable {
  public:
   virtual ~SecondaryCache() {}
 
-  virtual std::string Name() = 0;
-
-  static const std::string Type() { return "SecondaryCache"; }
+  static const char* Type() { return "SecondaryCache"; }
+  static Status CreateFromString(const ConfigOptions& config_options,
+                                 const std::string& id,
+                                 std::shared_ptr<SecondaryCache>* result);
 
   // Insert the given value into this cache. The value is not written
   // directly. Rather, the SaveToCallback provided by helper_cb will be
diff --git a/dist/linux_amd64/include/rocksdb/statistics.h b/dist/linux_amd64/include/rocksdb/statistics.h
index 8fc5a2e..f5e83b3 100644
--- a/dist/linux_amd64/include/rocksdb/statistics.h
+++ b/dist/linux_amd64/include/rocksdb/statistics.h
@@ -578,7 +578,7 @@ class Statistics {
   // Resets all ticker and histogram stats
   virtual Status Reset() { return Status::NotSupported("Not implemented"); }
 
-  // String representation of the statistic object.
+  // String representation of the statistic object. Must be thread-safe.
   virtual std::string ToString() const {
     // Do nothing by default
     return std::string("ToString(): not implemented");
diff --git a/dist/linux_amd64/include/rocksdb/status.h b/dist/linux_amd64/include/rocksdb/status.h
index 1de2ebc..a60212b 100644
--- a/dist/linux_amd64/include/rocksdb/status.h
+++ b/dist/linux_amd64/include/rocksdb/status.h
@@ -452,10 +452,8 @@ class Status {
 
  protected:
   // A nullptr state_ (which is always the case for OK) means the message
-  // is empty.
-  // of the following form:
-  //    state_[0..3] == length of message
-  //    state_[4..]  == message
+  // is empty, else state_ points to message.
+
   Code code_;
   SubCode subcode_;
   Severity sev_;
diff --git a/dist/linux_amd64/include/rocksdb/utilities/customizable_util.h b/dist/linux_amd64/include/rocksdb/utilities/customizable_util.h
new file mode 100644
index 0000000..7d62921
--- /dev/null
+++ b/dist/linux_amd64/include/rocksdb/utilities/customizable_util.h
@@ -0,0 +1,279 @@
+// 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).
+
+#pragma once
+#include <functional>
+#include <memory>
+#include <unordered_map>
+
+#include "options/configurable_helper.h"
+#include "rocksdb/convenience.h"
+#include "rocksdb/customizable.h"
+#include "rocksdb/status.h"
+#include "rocksdb/utilities/object_registry.h"
+
+namespace ROCKSDB_NAMESPACE {
+// The FactoryFunc functions are used to create a new customizable object
+// without going through the ObjectRegistry.  This methodology is especially
+// useful in LITE mode, where there is no ObjectRegistry.  The methods take
+// in an ID of the object to create and a pointer to store the created object.
+// If the factory successfully recognized the input ID, the method should return
+// success; otherwise false should be returned.  On success, the object
+// parameter contains the new object.
+template <typename T>
+using SharedFactoryFunc =
+    std::function<bool(const std::string&, std::shared_ptr<T>*)>;
+
+template <typename T>
+using UniqueFactoryFunc =
+    std::function<bool(const std::string&, std::unique_ptr<T>*)>;
+
+template <typename T>
+using StaticFactoryFunc = std::function<bool(const std::string&, T**)>;
+
+// Creates a new shared customizable instance object based on the
+// input parameters using the object registry.
+//
+// The id parameter specifies the instance class of the object to create.
+// The opt_map parameter specifies the configuration of the new instance.
+//
+// The config_options parameter controls the process and how errors are
+// returned. If ignore_unknown_options=true, unknown values are ignored during
+// the configuration. If ignore_unsupported_options=true, unknown instance types
+// are ignored. If invoke_prepare_options=true, the resulting instance will be
+// initialized (via PrepareOptions)
+//
+// @param config_options Controls how the instance is created and errors are
+// handled
+// @param id The identifier of the new object being created.  This string
+// will be used by the object registry to locate the appropriate object to
+// create.
+// @param opt_map Optional name-value pairs of properties to set for the newly
+// created object
+// @param result The newly created and configured instance.
+template <typename T>
+static Status NewSharedObject(
+    const ConfigOptions& config_options, const std::string& id,
+    const std::unordered_map<std::string, std::string>& opt_map,
+    std::shared_ptr<T>* result) {
+  Status status;
+  if (!id.empty()) {
+#ifndef ROCKSDB_LITE
+    status = config_options.registry->NewSharedObject(id, result);
+#else
+    status = Status::NotSupported("Cannot load object in LITE mode ", id);
+#endif  // ROCKSDB_LITE
+    if (config_options.ignore_unsupported_options && status.IsNotSupported()) {
+      return Status::OK();
+    }
+  } else {
+    status = Status::NotSupported("Cannot reset object ");
+  }
+  if (!status.ok()) {
+    return status;
+  } else {
+    return Customizable::ConfigureNewObject(config_options, result->get(),
+                                            opt_map);
+  }
+}
+
+// Creates a new shared Customizable object based on the input parameters.
+// This method parses the input value to determine the type of instance to
+// create. If there is an existing instance (in result) and it is the same ID
+// as the object being created, the existing configuration is stored and used as
+// the default for the new object.
+//
+// The value parameter specified the instance class of the object to create.
+// If it is a simple string (e.g. BlockBasedTable), then the instance will be
+// created using the default settings.  If the value is a set of name-value
+// pairs, then the "id" value is used to determine the instance to create and
+// the remaining parameters are used to configure the object.  Id name-value
+// pairs are specified, there should be an "id=value" pairing or an error may
+// result.
+//
+// The config_options parameter controls the process and how errors are
+// returned. If ignore_unknown_options=true, unknown values are ignored during
+// the configuration. If ignore_unsupported_options=true, unknown instance types
+// are ignored. If invoke_prepare_options=true, the resulting instance will be
+// initialized (via PrepareOptions)
+//
+// @param config_options Controls how the instance is created and errors are
+// handled
+// @param value Either the simple name of the instance to create, or a set of
+// name-value pairs to create and initailize the object
+// @param func  Optional function to call to attempt to create an instance
+// @param result The newly created instance.
+template <typename T>
+static Status LoadSharedObject(const ConfigOptions& config_options,
+                               const std::string& value,
+                               const SharedFactoryFunc<T>& func,
+                               std::shared_ptr<T>* result) {
+  std::string id;
+  std::unordered_map<std::string, std::string> opt_map;
+
+  Status status = Customizable::GetOptionsMap(config_options, result->get(),
+                                              value, &id, &opt_map);
+  if (!status.ok()) {  // GetOptionsMap failed
+    return status;
+  } else if (func == nullptr ||
+             !func(id, result)) {  // No factory, or it failed
+    if (value.empty()) {           // No Id and no options.  Clear the object
+      *result = nullptr;
+      return Status::OK();
+    } else {
+      return NewSharedObject(config_options, id, opt_map, result);
+    }
+  } else {
+    return Customizable::ConfigureNewObject(config_options, result->get(),
+                                            opt_map);
+  }
+}
+
+// Creates a new unique pointer customizable instance object based on the
+// input parameters using the object registry.
+// @see NewSharedObject for more information on the inner workings of this
+// method.
+//
+// @param config_options Controls how the instance is created and errors are
+// handled
+// @param id The identifier of the new object being created.  This string
+// will be used by the object registry to locate the appropriate object to
+// create.
+// @param opt_map Optional name-value pairs of properties to set for the newly
+// created object
+// @param result The newly created and configured instance.
+template <typename T>
+static Status NewUniqueObject(
+    const ConfigOptions& config_options, const std::string& id,
+    const std::unordered_map<std::string, std::string>& opt_map,
+    std::unique_ptr<T>* result) {
+  Status status;
+  if (id.empty()) {
+    status = Status::NotSupported("Cannot reset object ");
+  } else {
+#ifndef ROCKSDB_LITE
+    status = config_options.registry->NewUniqueObject(id, result);
+#else
+    status = Status::NotSupported("Cannot load object in LITE mode ", id);
+#endif  // ROCKSDB_LITE
+    if (config_options.ignore_unsupported_options && status.IsNotSupported()) {
+      return Status::OK();
+    }
+  }
+  if (!status.ok()) {
+    return status;
+  } else {
+    return Customizable::ConfigureNewObject(config_options, result->get(),
+                                            opt_map);
+  }
+}
+
+// Creates a new unique customizable instance object based on the input
+// parameters.
+// @see LoadSharedObject for more information on the inner workings of this
+// method.
+//
+// @param config_options Controls how the instance is created and errors are
+// handled
+// @param value Either the simple name of the instance to create, or a set of
+// name-value pairs to create and initailize the object
+// @param func  Optional function to call to attempt to create an instance
+// @param result The newly created instance.
+template <typename T>
+static Status LoadUniqueObject(const ConfigOptions& config_options,
+                               const std::string& value,
+                               const UniqueFactoryFunc<T>& func,
+                               std::unique_ptr<T>* result) {
+  std::string id;
+  std::unordered_map<std::string, std::string> opt_map;
+  Status status = Customizable::GetOptionsMap(config_options, result->get(),
+                                              value, &id, &opt_map);
+  if (!status.ok()) {  // GetOptionsMap failed
+    return status;
+  } else if (func == nullptr ||
+             !func(id, result)) {  // No factory, or it failed
+    if (value.empty()) {           // No Id and no options.  Clear the object
+      *result = nullptr;
+      return Status::OK();
+    } else {
+      return NewUniqueObject(config_options, id, opt_map, result);
+    }
+  } else {
+    return Customizable::ConfigureNewObject(config_options, result->get(),
+                                            opt_map);
+  }
+}
+
+// Creates a new static (raw pointer) customizable instance object based on the
+// input parameters using the object registry.
+// @see NewSharedObject for more information on the inner workings of this
+// method.
+//
+// @param config_options Controls how the instance is created and errors are
+// handled
+// @param id The identifier of the new object being created.  This string
+// will be used by the object registry to locate the appropriate object to
+// create.
+// @param opt_map Optional name-value pairs of properties to set for the newly
+// created object
+// @param result The newly created and configured instance.
+template <typename T>
+static Status NewStaticObject(
+    const ConfigOptions& config_options, const std::string& id,
+    const std::unordered_map<std::string, std::string>& opt_map, T** result) {
+  Status status;
+  if (id.empty()) {
+    status = Status::NotSupported("Cannot reset object ");
+  } else {
+#ifndef ROCKSDB_LITE
+    status = config_options.registry->NewStaticObject(id, result);
+#else
+    status = Status::NotSupported("Cannot load object in LITE mode ", id);
+#endif  // ROCKSDB_LITE
+    if (config_options.ignore_unsupported_options && status.IsNotSupported()) {
+      return Status::OK();
+    }
+  }
+  if (!status.ok()) {
+    return status;
+  } else {
+    return Customizable::ConfigureNewObject(config_options, *result, opt_map);
+  }
+}
+
+// Creates a new static (raw pointer) customizable instance object based on the
+// input parameters.
+// @see LoadSharedObject for more information on the inner workings of this
+// method.
+//
+// @param config_options Controls how the instance is created and errors are
+// handled
+// @param value Either the simple name of the instance to create, or a set of
+// name-value pairs to create and initailize the object
+// @param func  Optional function to call to attempt to create an instance
+// @param result The newly created instance.
+template <typename T>
+static Status LoadStaticObject(const ConfigOptions& config_options,
+                               const std::string& value,
+                               const StaticFactoryFunc<T>& func, T** result) {
+  std::string id;
+  std::unordered_map<std::string, std::string> opt_map;
+  Status status = Customizable::GetOptionsMap(config_options, *result, value,
+                                              &id, &opt_map);
+  if (!status.ok()) {  // GetOptionsMap failed
+    return status;
+  } else if (func == nullptr ||
+             !func(id, result)) {  // No factory, or it failed
+    if (value.empty()) {           // No Id and no options.  Clear the object
+      *result = nullptr;
+      return Status::OK();
+    } else {
+      return NewStaticObject(config_options, id, opt_map, result);
+    }
+  } else {
+    return Customizable::ConfigureNewObject(config_options, *result, opt_map);
+  }
+}
+}  // namespace ROCKSDB_NAMESPACE
diff --git a/dist/linux_amd64/include/rocksdb/utilities/object_registry.h b/dist/linux_amd64/include/rocksdb/utilities/object_registry.h
index 5a454d7..946a267 100644
--- a/dist/linux_amd64/include/rocksdb/utilities/object_registry.h
+++ b/dist/linux_amd64/include/rocksdb/utilities/object_registry.h
@@ -9,10 +9,12 @@
 
 #include <functional>
 #include <memory>
+#include <mutex>
 #include <regex>
 #include <string>
 #include <unordered_map>
 #include <vector>
+
 #include "rocksdb/status.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -109,6 +111,8 @@ class ObjectLibrary {
   // Adds the input entry to the list for the given type
   void AddEntry(const std::string& type, std::unique_ptr<Entry>& entry);
 
+  // Protects the entry map
+  mutable std::mutex mu_;
   // ** FactoryFunctions for this loader, organized by type
   std::unordered_map<std::string, std::vector<std::unique_ptr<Entry>>> entries_;
 
diff --git a/dist/linux_amd64/include/rocksdb/utilities/options_type.h b/dist/linux_amd64/include/rocksdb/utilities/options_type.h
index 7057c78..bf96ec7 100644
--- a/dist/linux_amd64/include/rocksdb/utilities/options_type.h
+++ b/dist/linux_amd64/include/rocksdb/utilities/options_type.h
@@ -35,14 +35,12 @@ enum class OptionType {
   kCompactionPri,
   kSliceTransform,
   kCompressionType,
-  kComparator,
   kCompactionFilter,
   kCompactionFilterFactory,
   kCompactionStopStyle,
   kMergeOperator,
   kMemTableRepFactory,
   kFilterPolicy,
-  kFlushBlockPolicyFactory,
   kChecksumType,
   kEncodingType,
   kEnv,
diff --git a/dist/linux_amd64/include/rocksdb/version.h b/dist/linux_amd64/include/rocksdb/version.h
index ebdfb74..133b15f 100644
--- a/dist/linux_amd64/include/rocksdb/version.h
+++ b/dist/linux_amd64/include/rocksdb/version.h
@@ -10,8 +10,8 @@
 #include "rocksdb/rocksdb_namespace.h"
 
 #define ROCKSDB_MAJOR 6
-#define ROCKSDB_MINOR 22
-#define ROCKSDB_PATCH 1
+#define ROCKSDB_MINOR 23
+#define ROCKSDB_PATCH 3
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these
diff --git a/dist/linux_amd64/lib/liblz4.a b/dist/linux_amd64/lib/liblz4.a
index 65068d5..e646203 100644
Binary files a/dist/linux_amd64/lib/liblz4.a and b/dist/linux_amd64/lib/liblz4.a differ
diff --git a/dist/linux_amd64/lib/librocksdb.a b/dist/linux_amd64/lib/librocksdb.a
index 2d74da1..286062e 100644
Binary files a/dist/linux_amd64/lib/librocksdb.a and b/dist/linux_amd64/lib/librocksdb.a differ
diff --git a/dist/linux_amd64/lib/libsnappy.a b/dist/linux_amd64/lib/libsnappy.a
index 2506442..2facae6 100644
Binary files a/dist/linux_amd64/lib/libsnappy.a and b/dist/linux_amd64/lib/libsnappy.a differ
diff --git a/dist/linux_amd64/lib/libz.a b/dist/linux_amd64/lib/libz.a
index 36ed743..54d58ad 100644
Binary files a/dist/linux_amd64/lib/libz.a and b/dist/linux_amd64/lib/libz.a differ
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 8ee3593..db17708 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -21,7 +21,7 @@ ENV CFLAGS='-fPIC -O3 -pipe'
 ENV CXXFLAGS='-fPIC -O3 -pipe'
 
 # install go
-ENV GOLANG_PACKAGE go1.16.4.linux-amd64.tar.gz
+ENV GOLANG_PACKAGE go1.17.2.linux-amd64.tar.gz
 
 RUN curl https://dl.google.com/go/${GOLANG_PACKAGE} -o ${GOLANG_PACKAGE} && \
     tar -C /usr/local -xzf ${GOLANG_PACKAGE} && rm ${GOLANG_PACKAGE}
diff --git a/filter_policy.go b/filter_policy.go
index 79ba10e..bd9c68b 100644
--- a/filter_policy.go
+++ b/filter_policy.go
@@ -71,6 +71,31 @@ func NewBloomFilterFull(bitsPerKey int) FilterPolicy {
 	return NewNativeFilterPolicy(C.rocksdb_filterpolicy_create_bloom_full(C.int(bitsPerKey)))
 }
 
+// NewRibbonFilterPolicy create a new Bloom alternative that saves about 30% space compared to
+// Bloom filters, with similar query times but roughly 3-4x CPU time
+// and 3x temporary space usage during construction. For example, if
+// you pass in 10 for bloom_equivalent_bits_per_key, you'll get the same
+// 0.95% FP rate as Bloom filter but only using about 7 bits per key.
+//
+// Ribbon filters are compatible with RocksDB >= 6.15.0. Earlier
+// versions reading the data will behave as if no filter was used
+// (degraded performance until compaction rebuilds filters). All
+// built-in FilterPolicies (Bloom or Ribbon) are able to read other
+// kinds of built-in filters.
+//
+// Note: the current Ribbon filter schema uses some extra resources
+// when constructing very large filters. For example, for 100 million
+// keys in a single filter (one SST file without partitioned filters),
+// 3GB of temporary, untracked memory is used, vs. 1GB for Bloom.
+// However, the savings in filter space from just ~60 open SST files
+// makes up for the additional temporary memory use.
+//
+// Also consider using optimize_filters_for_memory to save filter
+// memory.
+func NewRibbonFilterPolicy(bloom_equivalent_bits_per_key int) FilterPolicy {
+	return NewNativeFilterPolicy(C.rocksdb_filterpolicy_create_ribbon(C.int(bloom_equivalent_bits_per_key)))
+}
+
 // Hold references to filter policies.
 var filterPolicies = NewCOWList()
 
diff --git a/go.mod b/go.mod
index a8e82a2..28cd343 100644
--- a/go.mod
+++ b/go.mod
@@ -1,7 +1,9 @@
 module github.com/linxGnu/grocksdb
 
+require github.com/stretchr/testify v1.7.0
+
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/stretchr/testify v1.7.0
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 )