perf: use unique_ptr instead of shared_ptr (#1374)

performance: replace unnecessary shared_ptr with unique_ptr to eliminate atomic refcounting and reduce contention on hot insert paths.

Author: gongxun0928

contrib/pax_storage/src/cpp/storage/columns/pax_column.cc

@@ -208,7 +208,7 @@ std::string PaxColumn::DebugString() {
 
 template <typename T>
 PaxCommColumn<T>::PaxCommColumn(uint32 capacity) {
-  data_ = std::make_shared<DataBuffer<T>>(capacity * sizeof(T));
+  data_ = std::make_unique<DataBuffer<T>>(capacity * sizeof(T));
 }
 
 template <typename T>
@@ -218,7 +218,7 @@ template <typename T>  // NOLINT: redirect constructor
 PaxCommColumn<T>::PaxCommColumn() : PaxCommColumn(DEFAULT_CAPACITY) {}
 
 template <typename T>
-void PaxCommColumn<T>::Set(std::shared_ptr<DataBuffer<T>> data) {
+void PaxCommColumn<T>::Set(std::unique_ptr<DataBuffer<T>> data) {
   data_ = std::move(data);
 }
 
@@ -318,17 +318,17 @@ template class PaxCommColumn<double>;
 PaxNonFixedColumn::PaxNonFixedColumn(uint32 data_capacity,
                                      uint32 offsets_capacity)
     : estimated_size_(0),
-      data_(std::make_shared<DataBuffer<char>>(data_capacity)),
-      offsets_(std::make_shared<DataBuffer<int32>>(offsets_capacity)),
+      data_(std::make_unique<DataBuffer<char>>(data_capacity)),
+      offsets_(std::make_unique<DataBuffer<int32>>(offsets_capacity)),
       next_offsets_(0) {}
 
 PaxNonFixedColumn::PaxNonFixedColumn()
     : PaxNonFixedColumn(DEFAULT_CAPACITY, DEFAULT_CAPACITY) {}
 
 PaxNonFixedColumn::~PaxNonFixedColumn() {}
 
-void PaxNonFixedColumn::Set(std::shared_ptr<DataBuffer<char>> data,
-                            std::shared_ptr<DataBuffer<int32>> offsets,
+void PaxNonFixedColumn::Set(std::unique_ptr<DataBuffer<char>> data,
+                            std::unique_ptr<DataBuffer<int32>> offsets,
                             size_t total_size) {
   estimated_size_ = total_size;
   data_ = std::move(data);

contrib/pax_storage/src/cpp/storage/columns/pax_column.h

@@ -239,7 +239,9 @@ class PaxColumn {
   }
 
   // Get the null bitmap
-  inline const std::unique_ptr<Bitmap8>& GetBitmap() const { return null_bitmap_; }
+  inline const std::unique_ptr<Bitmap8> &GetBitmap() const {
+    return null_bitmap_;
+  }
 
   // Set the column kv attributes
   void SetAttributes(const std::map<std::string, std::string> &attrs);
@@ -425,7 +427,7 @@ class PaxCommColumn : public PaxColumn {
 
   PaxCommColumn();
 
-  virtual void Set(std::shared_ptr<DataBuffer<T>> data);
+  virtual void Set(std::unique_ptr<DataBuffer<T>> data);
 
   PaxColumnTypeInMem GetPaxColumnTypeInMem() const override;
 
@@ -455,7 +457,7 @@ class PaxCommColumn : public PaxColumn {
   int32 GetTypeLength() const override;
 
  protected:
-  std::shared_ptr<DataBuffer<T>> data_;
+  std::unique_ptr<DataBuffer<T>> data_;
 };
 
 extern template class PaxCommColumn<char>;
@@ -474,8 +476,8 @@ class PaxNonFixedColumn : public PaxColumn {
 
   ~PaxNonFixedColumn() override;
 
-  virtual void Set(std::shared_ptr<DataBuffer<char>> data,
-                   std::shared_ptr<DataBuffer<int32>> offsets,
+  virtual void Set(std::unique_ptr<DataBuffer<char>> data,
+                   std::unique_ptr<DataBuffer<int32>> offsets,
                    size_t total_size);
 
   void Append(char *buffer, size_t size) override;
@@ -514,13 +516,13 @@ class PaxNonFixedColumn : public PaxColumn {
 
  protected:
   size_t estimated_size_;
-  std::shared_ptr<DataBuffer<char>> data_;
+  std::unique_ptr<DataBuffer<char>> data_;
 
   // orc needs to serialize int32 array
   // the length of a single tuple field will not exceed 2GB,
   // so a variable-length element of the offsets stream can use int32
   // to represent the length
-  std::shared_ptr<DataBuffer<int32>> offsets_;
+  std::unique_ptr<DataBuffer<int32>> offsets_;
 
   // used to record next offset in write path
   // in read path, next_offsets_ always be -1

contrib/pax_storage/src/cpp/storage/columns/pax_column_test.cc

@@ -137,61 +137,61 @@ static std::unique_ptr<PaxColumn> CreateDecodeColumn(
   std::unique_ptr<PaxColumn> column_rc;
   switch (bits) {
     case 16: {
-      auto buffer_for_read = std::make_shared<DataBuffer<int16>>(
+      auto buffer_for_read = std::make_unique<DataBuffer<int16>>(
           reinterpret_cast<int16 *>(encoded_buff), encoded_len, false, false);
       buffer_for_read->Brush(encoded_len);
 
       if (storage_type == PaxStorageFormat::kTypeStoragePorcNonVec) {
         auto int_column =
             ColumnOptCreateTraits<PaxEncodingColumn, int16>::create_decoding(
                 origin_len / sizeof(int16), std::move(decoding_option));
-        int_column->Set(buffer_for_read);
+        int_column->Set(std::move(buffer_for_read));
         column_rc = std::move(int_column);
       } else {
         auto int_column =
             ColumnOptCreateTraits<PaxVecEncodingColumn, int16>::create_decoding(
                 origin_len / sizeof(int16), std::move(decoding_option));
-        int_column->Set(buffer_for_read, column_not_nulls);
+        int_column->Set(std::move(buffer_for_read), column_not_nulls);
         column_rc = std::move(int_column);
       }
       break;
     }
     case 32: {
-      auto buffer_for_read = std::make_shared<DataBuffer<int32>>(
+      auto buffer_for_read = std::make_unique<DataBuffer<int32>>(
           reinterpret_cast<int32 *>(encoded_buff), encoded_len, false, false);
       buffer_for_read->Brush(encoded_len);
 
       if (storage_type == PaxStorageFormat::kTypeStoragePorcNonVec) {
         auto int_column =
             ColumnOptCreateTraits<PaxEncodingColumn, int32>::create_decoding(
                 origin_len / sizeof(int32), std::move(decoding_option));
-        int_column->Set(buffer_for_read);
+        int_column->Set(std::move(buffer_for_read));
         column_rc = std::move(int_column);
       } else {
         auto int_column =
             ColumnOptCreateTraits<PaxVecEncodingColumn, int32>::create_decoding(
                 origin_len / sizeof(int32), std::move(decoding_option));
-        int_column->Set(buffer_for_read, column_not_nulls);
+        int_column->Set(std::move(buffer_for_read), column_not_nulls);
         column_rc = std::move(int_column);
       }
       break;
     }
     case 64: {
-      auto buffer_for_read = std::make_shared<DataBuffer<int64>>(
+      auto buffer_for_read = std::make_unique<DataBuffer<int64>>(
           reinterpret_cast<int64 *>(encoded_buff), encoded_len, false, false);
       buffer_for_read->Brush(encoded_len);
 
       if (storage_type == PaxStorageFormat::kTypeStoragePorcNonVec) {
         auto int_column =
             ColumnOptCreateTraits<PaxEncodingColumn, int64>::create_decoding(
                 origin_len / sizeof(int64), std::move(decoding_option));
-        int_column->Set(buffer_for_read);
+        int_column->Set(std::move(buffer_for_read));
         column_rc = std::move(int_column);
       } else {
         auto int_column =
             ColumnOptCreateTraits<PaxVecEncodingColumn, int64>::create_decoding(
                 origin_len / sizeof(int64), std::move(decoding_option));
-        int_column->Set(buffer_for_read, column_not_nulls);
+        int_column->Set(std::move(buffer_for_read), column_not_nulls);
         column_rc = std::move(int_column);
       }
       break;
@@ -749,14 +749,14 @@ TEST_P(PaxNonFixedColumnCompressTest,
   auto non_fixed_column_for_read = new PaxNonFixedEncodingColumn(
       number_of_rows * number, sizeof(int32) * number_of_rows,
       std::move(decoding_option));
-  auto data_buffer_for_read = std::make_shared<DataBuffer<char>>(
+  auto data_buffer_for_read = std::make_unique<DataBuffer<char>>(
       encoded_buff, encoded_len, false, false);
   data_buffer_for_read->Brush(encoded_len);
-  auto length_buffer_cpy = std::make_shared<DataBuffer<int32>>(
+  auto length_buffer_cpy = std::make_unique<DataBuffer<int32>>(
       (int32 *)offset_stream_buff, offset_stream_len, false, false);
   length_buffer_cpy->BrushAll();
-  non_fixed_column_for_read->Set(data_buffer_for_read, length_buffer_cpy,
-                                 origin_len);
+  non_fixed_column_for_read->Set(std::move(data_buffer_for_read),
+                                 std::move(length_buffer_cpy), origin_len);
   ASSERT_EQ(non_fixed_column_for_read->GetCompressLevel(), 5);
   char *verify_buff;
   size_t verify_len;

contrib/pax_storage/src/cpp/storage/columns/pax_encoding_column.cc

@@ -124,7 +124,7 @@ void PaxEncodingColumn<T>::InitDecoder() {
 }
 
 template <typename T>
-void PaxEncodingColumn<T>::Set(std::shared_ptr<DataBuffer<T>> data) {
+void PaxEncodingColumn<T>::Set(std::unique_ptr<DataBuffer<T>> data) {
   if (decoder_) {
     // should not decoding null
     if (data->Used() != 0) {
@@ -155,7 +155,7 @@ void PaxEncodingColumn<T>::Set(std::shared_ptr<DataBuffer<T>> data) {
 
     Assert(!data->IsMemTakeOver());
   } else {
-    PaxCommColumn<T>::Set(data);
+    PaxCommColumn<T>::Set(std::move(data));
   }
 }
 
@@ -175,7 +175,7 @@ std::pair<char *, size_t> PaxEncodingColumn<T>::GetBuffer() {
     if (encoder_) {
       // changed streaming encode to blocking encode
       // because we still need store a origin data in `PaxCommColumn<T>`
-      auto origin_data_buffer = PaxCommColumn<T>::data_;
+      auto origin_data_buffer = PaxCommColumn<T>::data_.get();
 
       shared_data_ = std::make_shared<DataBuffer<char>>(origin_data_buffer->Used());
       encoder_->SetDataBuffer(shared_data_);

contrib/pax_storage/src/cpp/storage/columns/pax_encoding_column.h

@@ -44,7 +44,7 @@ class PaxEncodingColumn : public PaxCommColumn<T> {
 
   ~PaxEncodingColumn() override;
 
-  void Set(std::shared_ptr<DataBuffer<T>> data) override;
+  void Set(std::unique_ptr<DataBuffer<T>> data) override;
 
   std::pair<char *, size_t> GetBuffer() override;
 

contrib/pax_storage/src/cpp/storage/columns/pax_encoding_non_fixed_column.cc

@@ -145,8 +145,8 @@ PaxNonFixedEncodingColumn::PaxNonFixedEncodingColumn(
 
 PaxNonFixedEncodingColumn::~PaxNonFixedEncodingColumn() {}
 
-void PaxNonFixedEncodingColumn::Set(std::shared_ptr<DataBuffer<char>> data,
-                                    std::shared_ptr<DataBuffer<int32>> offsets,
+void PaxNonFixedEncodingColumn::Set(std::unique_ptr<DataBuffer<char>> data,
+                                    std::unique_ptr<DataBuffer<int32>> offsets,
                                     size_t total_size) {
   bool exist_decoder;
   Assert(data && offsets);
@@ -179,7 +179,7 @@ void PaxNonFixedEncodingColumn::Set(std::shared_ptr<DataBuffer<char>> data,
       // `data_` have the same buffer with `shared_data_`
       PaxNonFixedColumn::data_->Brush(shared_data_->Used());
       // no delete the origin data
-      shared_data_ = data;
+      shared_data_ = std::move(data);
     }
   };
 
@@ -228,16 +228,16 @@ void PaxNonFixedEncodingColumn::Set(std::shared_ptr<DataBuffer<char>> data,
     PaxNonFixedColumn::next_offsets_ = -1;
   } else if (exist_decoder && !has_offsets_processor) {
     data_decompress();
-    PaxNonFixedColumn::offsets_ = offsets;
+    PaxNonFixedColumn::offsets_ = std::move(offsets);
     PaxNonFixedColumn::estimated_size_ = total_size;
     PaxNonFixedColumn::next_offsets_ = -1;
   } else if (!exist_decoder && has_offsets_processor) {
-    PaxNonFixedColumn::data_ = data;
+    PaxNonFixedColumn::data_ = std::move(data);
     offsets_decompress();
     PaxNonFixedColumn::estimated_size_ = total_size;
     PaxNonFixedColumn::next_offsets_ = -1;
   } else {  // (!compressor_ && !offsets_compressor_)
-    PaxNonFixedColumn::Set(data, offsets, total_size);
+    PaxNonFixedColumn::Set(std::move(data), std::move(offsets), total_size);
   }
 }
 

contrib/pax_storage/src/cpp/storage/columns/pax_encoding_non_fixed_column.h

@@ -42,8 +42,8 @@ class PaxNonFixedEncodingColumn : public PaxNonFixedColumn {
 
   ~PaxNonFixedEncodingColumn() override;
 
-  void Set(std::shared_ptr<DataBuffer<char>> data,
-           std::shared_ptr<DataBuffer<int32>> offsets,
+  void Set(std::unique_ptr<DataBuffer<char>> data,
+           std::unique_ptr<DataBuffer<int32>> offsets,
            size_t total_size) override;
 
   std::pair<char *, size_t> GetBuffer() override;