[Feature] support column zero copy read from page cache

Signed-off-by: stdpain <drfeng08@gmail.com>

Author: stdpain

be/src/agent/publish_version.cpp

@@ -21,6 +21,7 @@
 #include "fmt/format.h"
 #include "gen_cpp/AgentService_types.h"
 #include "gutil/strings/join.h"
+#include "runtime/exec_env.h"
 #include "storage/data_dir.h"
 #include "storage/replication_txn_manager.h"
 #include "storage/storage_engine.h"

be/src/cache/datacache.cpp

@@ -22,6 +22,7 @@
 #include "common/status.h"
 #include "gutil/strings/split.h"
 #include "gutil/strings/strip.h"
+#include "runtime/exec_env.h"
 #include "storage/options.h"
 #include "util/parse_util.h"
 

be/src/cache/lrucache_engine.h

@@ -23,7 +23,7 @@ namespace starrocks {
 class LRUCacheEngine final : public LocalCacheEngine {
 public:
     LRUCacheEngine() = default;
-    virtual ~LRUCacheEngine() override = default;
+    ~LRUCacheEngine() override = default;
 
     Status init(const CacheOptions& options) override;
     bool is_initialized() const override { return _initialized.load(std::memory_order_relaxed); }

be/src/cache/object_cache/page_cache.cpp

@@ -44,9 +44,13 @@
 
 namespace starrocks {
 
+std::atomic<size_t> StoragePageCacheMetrics::returned_page_handle_count{};
+std::atomic<size_t> StoragePageCacheMetrics::released_page_handle_count{};
+
 METRIC_DEFINE_UINT_GAUGE(page_cache_lookup_count, MetricUnit::OPERATIONS);
 METRIC_DEFINE_UINT_GAUGE(page_cache_hit_count, MetricUnit::OPERATIONS);
 METRIC_DEFINE_UINT_GAUGE(page_cache_capacity, MetricUnit::BYTES);
+METRIC_DEFINE_UINT_GAUGE(page_cache_pinned_count, MetricUnit::BYTES);
 
 void StoragePageCache::init_metrics() {
     StarRocksMetrics::instance()->metrics()->register_metric("page_cache_lookup_count", &page_cache_lookup_count);
@@ -60,6 +64,10 @@ void StoragePageCache::init_metrics() {
     StarRocksMetrics::instance()->metrics()->register_metric("page_cache_capacity", &page_cache_capacity);
     StarRocksMetrics::instance()->metrics()->register_hook("page_cache_capacity",
                                                            [this]() { page_cache_capacity.set_value(get_capacity()); });
+
+    StarRocksMetrics::instance()->metrics()->register_metric("page_cache_pinned_count", &page_cache_pinned_count);
+    StarRocksMetrics::instance()->metrics()->register_hook(
+            "page_cache_pinned_count", [this]() { page_cache_pinned_count.set_value(get_pinned_count()); });
 }
 
 void StoragePageCache::prune() {
@@ -92,12 +100,17 @@ bool StoragePageCache::adjust_capacity(int64_t delta, size_t min_capacity) {
     return true;
 }
 
+size_t StoragePageCache::get_pinned_count() const {
+    return StoragePageCacheMetrics::returned_page_handle_count - StoragePageCacheMetrics::released_page_handle_count;
+}
+
 bool StoragePageCache::lookup(const std::string& key, PageCacheHandle* handle) {
     ObjectCacheHandle* obj_handle = nullptr;
     Status st = _cache->lookup(key, &obj_handle);
     if (!st.ok()) {
         return false;
     }
+    StoragePageCacheMetrics::returned_page_handle_count++;
     *handle = PageCacheHandle(_cache, obj_handle);
     return true;
 }
@@ -124,6 +137,7 @@ Status StoragePageCache::insert(const std::string& key, std::vector<uint8_t>* da
     Status st = _cache->insert(key, (void*)data, mem_size, deleter, &obj_handle, opts);
     if (st.ok()) {
         *handle = PageCacheHandle(_cache, obj_handle);
+        StoragePageCacheMetrics::returned_page_handle_count++;
     }
     return st;
 }
@@ -134,6 +148,7 @@ Status StoragePageCache::insert(const std::string& key, void* data, int64_t size
     Status st = _cache->insert(key, data, size, deleter, &obj_handle, opts);
     if (st.ok()) {
         *handle = PageCacheHandle(_cache, obj_handle);
+        StoragePageCacheMetrics::returned_page_handle_count++;
     }
     return st;
 }

be/src/cache/object_cache/page_cache.h

@@ -34,25 +34,25 @@
 
 #pragma once
 
-#include <memory>
 #include <string>
 #include <utility>
 
 #include "cache/datacache.h"
-#include "gutil/macros.h" // for DISALLOW_COPY
-#include "runtime/current_thread.h"
-#include "runtime/exec_env.h"
-#include "util/defer_op.h"
 
 namespace starrocks {
 
 class PageCacheHandle;
 class MemTracker;
-class ObjectCacheWriteOptions;
+struct ObjectCacheWriteOptions;
 
 // Page cache min size is 256MB
 static constexpr int64_t kcacheMinSize = 268435456;
 
+struct StoragePageCacheMetrics {
+    static std::atomic<size_t> returned_page_handle_count;
+    static std::atomic<size_t> released_page_handle_count;
+};
+
 // Wrapper around Cache, and used for cache page of column datas in Segment.
 // TODO(zc): We should add some metric to see cache hit/miss rate.
 class StoragePageCache {
@@ -110,6 +110,10 @@ class StoragePageCache {
     bool is_initialized() const { return _initialized.load(std::memory_order_relaxed); }
     bool available() const { return is_initialized() && _cache->mem_cache_available(); }
 
+    // get the number of pinned pages in the page cache
+    // used for metrics
+    size_t get_pinned_count() const;
+
 private:
     LocalCacheEngine* _cache = nullptr;
     std::atomic<bool> _initialized = false;
@@ -124,6 +128,7 @@ class PageCacheHandle {
     PageCacheHandle(LocalCacheEngine* cache, ObjectCacheHandle* handle) : _cache(cache), _handle(handle) {}
     ~PageCacheHandle() {
         if (_handle != nullptr) {
+            StoragePageCacheMetrics::released_page_handle_count++;
             _cache->release(_handle);
         }
     }

be/src/column/adaptive_nullable_column.cpp

@@ -35,7 +35,7 @@ size_t AdaptiveNullableColumn::null_count() const {
         if (!_has_null) {
             return 0;
         }
-        return SIMD::count_nonzero(_null_column->get_data());
+        return SIMD::count_nonzero(_null_column->immutable_data());
     }
     }
 }
@@ -53,7 +53,7 @@ size_t AdaptiveNullableColumn::null_count(size_t offset, size_t count) const {
         if (!_has_null) {
             return 0;
         }
-        return SIMD::count_nonzero(_null_column->get_data());
+        return SIMD::count_nonzero(_null_column->immutable_data());
     }
     }
 }
@@ -271,7 +271,7 @@ void AdaptiveNullableColumn::serialize_batch(uint8_t* dst, Buffer<uint32_t>& sli
                                              uint32_t max_one_row_size) const {
     materialized_nullable();
     _data_column->serialize_batch_with_null_masks(dst, slice_sizes, chunk_size, max_one_row_size,
-                                                  _null_column->get_data().data(), _has_null);
+                                                  _null_column->immutable_data().data(), _has_null);
 }
 
 const uint8_t* AdaptiveNullableColumn::deserialize_and_append(const uint8_t* pos) {

be/src/column/adaptive_nullable_column.h

@@ -86,15 +86,15 @@ class AdaptiveNullableColumn final
 
     AdaptiveNullableColumn(const AdaptiveNullableColumn& rhs) { CHECK(false) << "unimplemented"; }
 
-    AdaptiveNullableColumn(AdaptiveNullableColumn&& rhs) { CHECK(false) << "unimplemented"; }
+    AdaptiveNullableColumn(AdaptiveNullableColumn&& rhs) noexcept { CHECK(false) << "unimplemented"; }
 
     AdaptiveNullableColumn& operator=(const AdaptiveNullableColumn& rhs) {
         AdaptiveNullableColumn tmp(rhs);
         this->swap_column(tmp);
         return *this;
     }
 
-    AdaptiveNullableColumn& operator=(AdaptiveNullableColumn&& rhs) {
+    AdaptiveNullableColumn& operator=(AdaptiveNullableColumn&& rhs) noexcept {
         AdaptiveNullableColumn tmp(std::move(rhs));
         this->swap_column(tmp);
         return *this;
@@ -160,7 +160,7 @@ class AdaptiveNullableColumn final
             return false;
         }
         case State::kMaterialized: {
-            return _has_null && _null_column->get_data()[index];
+            return _has_null && _null_column->immutable_data()[index];
         }
         default: {
             __builtin_unreachable();
@@ -346,7 +346,7 @@ class AdaptiveNullableColumn final
 
     uint32_t serialize_size(size_t idx) const override {
         materialized_nullable();
-        if (_null_column->get_data()[idx]) {
+        if (_null_column->immutable_data()[idx]) {
             return sizeof(uint8_t);
         }
         return sizeof(uint8_t) + _data_column->serialize_size(idx);
@@ -458,10 +458,10 @@ class AdaptiveNullableColumn final
     // however, this may is not user want because once adaptive nullable column materialized,
     // its performance will be degraded to nullable column. Due to the following reason, we add
     // DCHECK(false) here and disable the behaviour.
-    const NullData& immutable_null_column_data() const {
+    const ImmutableNullData immutable_null_column_data() const {
         DCHECK(false);
         materialized_nullable();
-        return _null_column->get_data();
+        return _null_column->immutable_data();
     }
 
     Column* mutable_data_column() {