Fix metadata buffer leak in array-open

This fixes the following within StorageManager::load_array_metadata():
1. A direct memory leak of a `Buffer` object introduced with the chunked buffers
  patch.
2. An long-standing indirect memory leak of the metadata buffers.

For #1, an auxiliary `Buffer` object is created to store the contents of a
read from a chunked buffer. The `Buffer` object is never deleted.

For #2, the raw buffers inside of the `Buffer` object are passed into a
`ConstBuffer`. However, the `ConstBuffer` does not take ownership or free them.
The buffers are never freed while the `ConstBuffer` references them because of
leak #1. For this patch, I've replaced the `ConstBuffer` objects with `Buffer`
objects so that the `OpenArray` instance can take ownership of the metadata
buffers and free them appropriately.

Note that before the chunked buffer patch, we had a similar pattern where the
`Tile` buffer was disowned, but never freed by the `ConstBuffer`. This may be
affecting older core builds.

tiledb/sm/metadata/metadata.cc

@@ -89,7 +89,7 @@ Status Metadata::generate_uri(const URI& array_uri) {
 }
 
 Status Metadata::deserialize(
-    const std::vector<std::shared_ptr<ConstBuffer>>& metadata_buffs) {
+    const std::vector<std::shared_ptr<Buffer>>& metadata_buffs) {
   clear();
 
   if (metadata_buffs.empty())

tiledb/sm/metadata/metadata.h

@@ -108,7 +108,7 @@ class Metadata {
    * deleted or overwritten metadata items considering the order.
    */
   Status deserialize(
-      const std::vector<std::shared_ptr<ConstBuffer>>& metadata_buffs);
+      const std::vector<std::shared_ptr<Buffer>>& metadata_buffs);
 
   /** Serializes all key-value metadata items into the input buffer. */
   Status serialize(Buffer* buff) const;

tiledb/sm/storage_manager/open_array.cc

@@ -116,7 +116,7 @@ FragmentMetadata* OpenArray::fragment_metadata(const URI& uri) const {
   return (it == fragment_metadata_set_.end()) ? nullptr : it->second;
 }
 
-std::shared_ptr<ConstBuffer> OpenArray::array_metadata(const URI& uri) const {
+std::shared_ptr<Buffer> OpenArray::array_metadata(const URI& uri) const {
   std::lock_guard<std::mutex> lock(local_mtx_);
   auto it = array_metadata_.find(uri.to_string());
   return (it == array_metadata_.end()) ? nullptr : it->second;
@@ -146,7 +146,7 @@ void OpenArray::insert_fragment_metadata(FragmentMetadata* metadata) {
 }
 
 void OpenArray::insert_array_metadata(
-    const URI& uri, const std::shared_ptr<ConstBuffer>& metadata) {
+    const URI& uri, const std::shared_ptr<Buffer>& metadata) {
   std::lock_guard<std::mutex> lock(local_mtx_);
   assert(metadata != nullptr);
   array_metadata_[uri.to_string()] = metadata;

tiledb/sm/storage_manager/open_array.h

@@ -49,7 +49,7 @@ namespace tiledb {
 namespace sm {
 
 class ArraySchema;
-class ConstBuffer;
+class Buffer;
 class VFS;
 
 enum class QueryType : uint8_t;
@@ -127,7 +127,7 @@ class OpenArray {
    * Returns the constant buffer storing the serialized array metadata
    * of the input URI, or `nullptr` if the array metadata do not exist.
    */
-  std::shared_ptr<ConstBuffer> array_metadata(const URI& uri) const;
+  std::shared_ptr<Buffer> array_metadata(const URI& uri) const;
 
   /** Locks the array mutex. */
   void mtx_lock();
@@ -150,7 +150,7 @@ class OpenArray {
    * buffer) with the input URI.
    */
   void insert_array_metadata(
-      const URI& uri, const std::shared_ptr<ConstBuffer>& metadata);
+      const URI& uri, const std::shared_ptr<Buffer>& metadata);
 
   /** Sets an array schema. */
   void set_array_schema(ArraySchema* array_schema);
@@ -200,7 +200,7 @@ class OpenArray {
    * A map of URI strings to array metadata. The map stores the serialized
    * (decompressed, decrypted) array metadata into constant buffers.
    */
-  std::unordered_map<std::string, std::shared_ptr<ConstBuffer>> array_metadata_;
+  std::unordered_map<std::string, std::shared_ptr<Buffer>> array_metadata_;
 
   /**
    * A mutex used to lock the array for thread-safe open/close of the array

tiledb/sm/storage_manager/storage_manager.cc

@@ -2067,7 +2067,7 @@ Status StorageManager::load_array_metadata(
     return Status::Ok();
 
   auto metadata_num = array_metadata_to_load.size();
-  std::vector<std::shared_ptr<ConstBuffer>> metadata_buffs;
+  std::vector<std::shared_ptr<Buffer>> metadata_buffs;
   metadata_buffs.resize(metadata_num);
   auto statuses = parallel_for(0, metadata_num, [&](size_t m) {
     const auto& uri = array_metadata_to_load[m].uri_;
@@ -2078,17 +2078,14 @@ Status StorageManager::load_array_metadata(
       RETURN_NOT_OK(tile_io.read_generic(&tile, 0, encryption_key, config_));
 
       auto chunked_buffer = tile->chunked_buffer();
-      Buffer* const buff = new Buffer();
-      RETURN_NOT_OK(buff->realloc(chunked_buffer->size()));
-      buff->set_size(chunked_buffer->size());
+      metadata_buff = std::make_shared<Buffer>();
+      RETURN_NOT_OK(metadata_buff->realloc(chunked_buffer->size()));
+      metadata_buff->set_size(chunked_buffer->size());
       RETURN_NOT_OK_ELSE(
-          chunked_buffer->read(buff->data(), buff->size(), 0), [&]() {
-            delete tile;
-            delete buff;
-          }());
+          chunked_buffer->read(metadata_buff->data(), metadata_buff->size(), 0),
+          [&]() { delete tile; }());
       delete tile;
 
-      metadata_buff = std::make_shared<ConstBuffer>(buff);
       open_array->insert_array_metadata(uri, metadata_buff);
     }
     metadata_buffs[m] = metadata_buff;
@@ -2103,7 +2100,8 @@ Status StorageManager::load_array_metadata(
     meta_size += b->size();
   STATS_ADD_COUNTER(stats::Stats::CounterType::READ_ARRAY_META_SIZE, meta_size);
 
-  // Deserialize metadata buffers
+  // Deserialize metadata buffers. We must wrap the `Buffer` instances
+  // as `ConstBuffer` to conform to the interface.
   metadata->deserialize(metadata_buffs);
 
   // Sets the loaded metadata URIs