feat: hybrid column data cache (#17771)

* feat: hybrid disk cache

* cleanup

* fix typos

* tweak ut

* refine doc

* refactor: allow in-mem cache or hybrid cache to be none

* update ut test golden file

Author: dantengsky

Cargo.lock

@@ -3039,6 +3039,14 @@ pub struct CacheConfig {
     )]
     pub table_data_cache_population_queue_size: u32,
 
+    /// Bytes of data cache in-memory
+    #[clap(
+        long = "cache-data-cache-in-memory-bytes",
+        value_name = "VALUE",
+        default_value = "0"
+    )]
+    pub data_cache_in_memory_bytes: u64,
+
     /// Storage that hold the data caches
     #[clap(flatten)]
     #[serde(rename = "disk")]
@@ -3144,10 +3152,17 @@ pub struct DiskCacheConfig {
     )]
     pub path: String,
 
-    /// Whether sync data after write.
-    /// If the query node's memory is managed by cgroup (at least cgroup v1),
-    /// it's recommended to set this to true to prevent the container from
-    /// being killed due to high dirty page memory usage.
+    /// Controls whether to synchronize data to disk after write operations.
+    ///
+    /// When enabled, this option forces written data to be flushed to physical storage,
+    /// reducing the amount of dirty pages in memory. This is particularly important in
+    /// containerized environments where:
+    ///
+    /// 1. Memory limits are enforced by cgroups (v1, maybe v2 as well, though accounting differs)
+    /// 2. Dirty pages are counted against the memory limit, which increases the possibility of triggering OOM kills
+    ///
+    /// Setting this to true improves stability in memory-constrained / containerized environments at the cost
+    /// of potentially reduced write performance.
     #[clap(
         long = "cache-disk-sync-data",
         value_name = "VALUE",
@@ -3300,6 +3315,7 @@ mod cache_config_converters {
                 data_cache_storage: value.data_cache_storage.try_into()?,
                 table_data_cache_population_queue_size: value
                     .table_data_cache_population_queue_size,
+                data_cache_in_memory_bytes: value.data_cache_in_memory_bytes,
                 disk_cache_config: value.disk_cache_config.try_into()?,
                 data_cache_key_reload_policy: value.data_cache_key_reload_policy.try_into()?,
                 table_data_deserialized_data_bytes: value.table_data_deserialized_data_bytes,
@@ -3335,6 +3351,7 @@ mod cache_config_converters {
                 data_cache_key_reload_policy: value.data_cache_key_reload_policy.into(),
                 table_data_cache_population_queue_size: value
                     .table_data_cache_population_queue_size,
+                data_cache_in_memory_bytes: value.data_cache_in_memory_bytes,
                 disk_cache_config: value.disk_cache_config.into(),
                 table_data_deserialized_data_bytes: value.table_data_deserialized_data_bytes,
                 table_data_deserialized_memory_ratio: value.table_data_deserialized_memory_ratio,

src/query/config/src/config.rs

@@ -615,6 +615,9 @@ pub struct CacheConfig {
     ///   increase this value.
     pub table_data_cache_population_queue_size: u32,
 
+    /// Bytes of data cache in-memory
+    pub data_cache_in_memory_bytes: u64,
+
     /// Storage that hold the raw data caches
     pub disk_cache_config: DiskCacheConfig,
 
@@ -732,6 +735,7 @@ impl Default for CacheConfig {
             table_prune_partitions_count: 256,
             data_cache_storage: Default::default(),
             table_data_cache_population_queue_size: 0,
+            data_cache_in_memory_bytes: 0,
             disk_cache_config: Default::default(),
             data_cache_key_reload_policy: Default::default(),
             table_data_deserialized_data_bytes: 0,

src/query/config/src/inner.rs

@@ -5,6 +5,7 @@ DB.Table: 'system'.'configs', Table: configs-table_id:1, ver:0, Engine: SystemCo
 | Column 0  | Column 1                                        | Column 2                                                                                                                                                                                          | Column 3 |
 +-----------+-------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+----------+
 | 'cache'   | 'block_meta_count'                              | '0'                                                                                                                                                                                               | ''       |
+| 'cache'   | 'data_cache_in_memory_bytes'                    | '0'                                                                                                                                                                                               | ''       |
 | 'cache'   | 'data_cache_key_reload_policy'                  | 'reset'                                                                                                                                                                                           | ''       |
 | 'cache'   | 'data_cache_storage'                            | 'none'                                                                                                                                                                                            | ''       |
 | 'cache'   | 'disk.max_bytes'                                | '21474836480'                                                                                                                                                                                     | ''       |

src/query/service/tests/it/storages/testdata/configs_table_basic.txt

@@ -0,0 +1,67 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use bytes::Bytes;
+pub use databend_common_catalog::plan::PartStatistics;
+pub use databend_common_catalog::plan::Partitions;
+pub use databend_common_catalog::table::Table;
+use databend_common_exception::ErrorCode;
+pub use databend_storages_common_index::filters::Xor8Filter;
+pub use databend_storages_common_index::BloomIndexMeta;
+pub use databend_storages_common_index::InvertedIndexFile;
+pub use databend_storages_common_index::InvertedIndexMeta;
+pub use databend_storages_common_table_meta::meta::column_oriented_segment::ColumnOrientedSegment;
+pub use databend_storages_common_table_meta::meta::BlockMeta;
+pub use databend_storages_common_table_meta::meta::CompactSegmentInfo;
+pub use databend_storages_common_table_meta::meta::SegmentInfo;
+pub use databend_storages_common_table_meta::meta::TableSnapshot;
+pub use databend_storages_common_table_meta::meta::TableSnapshotStatistics;
+pub use parquet::file::metadata::ParquetMetaData;
+
+use crate::HybridCache;
+
+pub struct ColumnData(Bytes);
+
+impl ColumnData {
+    pub fn from_merge_io_read_result(bytes: Bytes) -> Self {
+        // Bytes are from merge read result, may refer to a large chunk of memory;
+        // Caching this large buffer wastes memory, so we need to copy it out.
+        ColumnData(bytes.to_vec().into())
+    }
+
+    pub fn bytes(&self) -> Bytes {
+        self.0.clone()
+    }
+
+    pub fn size(&self) -> usize {
+        self.0.len()
+    }
+}
+
+pub type ColumnDataCache = HybridCache<ColumnData>;
+impl TryFrom<&ColumnData> for Vec<u8> {
+    type Error = ErrorCode;
+    fn try_from(value: &ColumnData) -> Result<Self, Self::Error> {
+        Ok(value.0.to_vec())
+    }
+}
+
+// copying should be avoided
+impl TryFrom<Bytes> for ColumnData {
+    type Error = ErrorCode;
+
+    fn try_from(value: Bytes) -> Result<Self, Self::Error> {
+        Ok(ColumnData(value))
+    }
+}

src/query/storages/common/cache/src/cache_items.rs

@@ -18,21 +18,8 @@ use std::time::Instant;
 
 use arrow::array::ArrayRef;
 use databend_common_cache::MemSized;
-use databend_common_catalog::plan::PartStatistics;
-use databend_common_catalog::plan::Partitions;
-use databend_common_catalog::table::Table;
-use databend_storages_common_index::filters::Xor8Filter;
-use databend_storages_common_index::BloomIndexMeta;
-use databend_storages_common_index::InvertedIndexFile;
-use databend_storages_common_index::InvertedIndexMeta;
-use databend_storages_common_table_meta::meta::column_oriented_segment::ColumnOrientedSegment;
-use databend_storages_common_table_meta::meta::BlockMeta;
-use databend_storages_common_table_meta::meta::CompactSegmentInfo;
-use databend_storages_common_table_meta::meta::SegmentInfo;
-use databend_storages_common_table_meta::meta::TableSnapshot;
-use databend_storages_common_table_meta::meta::TableSnapshotStatistics;
-use parquet::file::metadata::ParquetMetaData;
 
+pub use crate::cache_items::*;
 use crate::manager::CacheManager;
 use crate::providers::HybridCache;
 use crate::CacheAccessor;
@@ -266,6 +253,15 @@ impl From<BloomIndexMeta> for CacheValue<BloomIndexMeta> {
     }
 }
 
+impl From<ColumnData> for CacheValue<ColumnData> {
+    fn from(value: ColumnData) -> Self {
+        CacheValue {
+            mem_bytes: value.size(),
+            inner: Arc::new(value),
+        }
+    }
+}
+
 impl From<InvertedIndexMeta> for CacheValue<InvertedIndexMeta> {
     fn from(value: InvertedIndexMeta) -> Self {
         CacheValue {

src/query/storages/common/cache/src/caches.rs

@@ -17,19 +17,15 @@
 #![feature(assert_matches)]
 
 mod cache;
+mod cache_items;
 mod caches;
 mod manager;
 mod providers;
 mod read;
 mod temp_dir;
 
-pub use cache::CacheAccessor;
-pub use cache::Unit;
-pub use caches::BlockMetaCache;
-pub use caches::CacheValue;
-pub use caches::CachedObject;
-pub use caches::SegmentBlockMetasCache;
-pub use caches::SizedColumnArray;
+pub use cache::*;
+pub use caches::*;
 pub use manager::CacheManager;
 // Unfortunately, criterion benchmarks need it to be pub
 pub use providers::disk_cache::disk_cache_lru::read_cache_content;
@@ -44,11 +40,5 @@ pub use providers::LruDiskCache;
 pub use providers::LruDiskCacheBuilder;
 pub use providers::LruDiskCacheHolder;
 pub use providers::TableDataCacheKey;
-pub use read::CacheKey;
-pub use read::CachedReader;
-pub use read::HybridCacheReader;
-pub use read::InMemoryCacheReader;
-pub use read::InMemoryCacheTTLReader;
-pub use read::LoadParams;
-pub use read::Loader;
+pub use read::*;
 pub use temp_dir::*;