Merge pull request ClickHouse#77356 from rschu1ze/fixups-qc-cache

Follow-up for query condition cache

Author: rschu1ze

src/Common/ProfileEvents.cpp

@@ -75,10 +75,10 @@
     M(MarkCacheMisses, "Number of times an entry has not been found in the mark cache, so we had to load a mark file in memory, which is a costly operation, adding to query latency.", ValueType::Number) \
     M(PrimaryIndexCacheHits, "Number of times an entry has been found in the primary index cache, so we didn't have to load a index file.", ValueType::Number) \
     M(PrimaryIndexCacheMisses, "Number of times an entry has not been found in the primary index cache, so we had to load a index file in memory, which is a costly operation, adding to query latency.", ValueType::Number) \
-    M(QueryCacheHits, "Number of times a query result has been found in the query cache (and query computation was avoided). Only updated for SELECT queries with SETTING use_query_cache = 1.", ValueType::Number) \
-    M(QueryCacheMisses, "Number of times a query result has not been found in the query cache (and required query computation). Only updated for SELECT queries with SETTING use_query_cache = 1.", ValueType::Number) \
     M(QueryConditionCacheHits, "Number of times an entry has been found in the query condition cache (and reading of marks can be skipped). Only updated for SELECT queries with SETTING use_query_condition_cache = 1.", ValueType::Number) \
     M(QueryConditionCacheMisses, "Number of times an entry has not been found in the query condition cache (and reading of mark cannot be skipped). Only updated for SELECT queries with SETTING use_query_condition_cache = 1.", ValueType::Number) \
+    M(QueryCacheHits, "Number of times a query result has been found in the query cache (and query computation was avoided). Only updated for SELECT queries with SETTING use_query_cache = 1.", ValueType::Number) \
+    M(QueryCacheMisses, "Number of times a query result has not been found in the query cache (and required query computation). Only updated for SELECT queries with SETTING use_query_cache = 1.", ValueType::Number) \
     M(CreatedReadBufferOrdinary, "Number of times ordinary read buffer was created for reading data (while choosing among other read methods).", ValueType::Number) \
     M(CreatedReadBufferDirectIO, "Number of times a read buffer with O_DIRECT was created for reading data (while choosing among other read methods).", ValueType::Number) \
     M(CreatedReadBufferDirectIOFailed, "Number of times a read buffer with O_DIRECT was attempted to be created for reading data (while choosing among other read methods), but the OS did not allow it (due to lack of filesystem support or other reasons) and we fallen back to the ordinary reading method.", ValueType::Number) \

src/Interpreters/Cache/QueryConditionCache.cpp

@@ -1,5 +1,8 @@
 #include <Interpreters/Cache/QueryConditionCache.h>
-#include <Storages/MergeTree/MergeTreeData.h>
+#include <Common/ProfileEvents.h>
+#include <Common/SipHash.h>
+#include <Common/logger_useful.h>
+#include <IO/WriteHelpers.h>
 
 namespace ProfileEvents
 {
@@ -16,10 +19,10 @@ QueryConditionCache::QueryConditionCache(const String & cache_policy, size_t max
 }
 
 void QueryConditionCache::write(
-    const UUID & table_id, const String & part_name, size_t predicate_hash,
+    const UUID & table_id, const String & part_name, size_t condition_hash,
     const MarkRanges & mark_ranges, size_t marks_count, bool has_final_mark)
 {
-    Key key = {table_id, part_name, predicate_hash};
+    Key key = {table_id, part_name, condition_hash};
 
     auto load_func = [&](){ return std::make_shared<Entry>(marks_count); };
     auto [entry, inserted] = cache.getOrSet(key, load_func);
@@ -38,33 +41,33 @@ void QueryConditionCache::write(
 
         LOG_DEBUG(
             logger,
-            "{} entry for table_id: {}, part_name: {}, predicate_hash: {}, marks_count: {}, has_final_mark: {}, ranges: {}",
+            "{} entry for table_id: {}, part_name: {}, condition_hash: {}, marks_count: {}, has_final_mark: {}, ranges: {}",
             inserted ? "Inserted" : "Updated",
             table_id,
             part_name,
-            predicate_hash,
+            condition_hash,
             marks_count,
             has_final_mark,
             toString(mark_ranges));
     }
 }
 
-std::optional<QueryConditionCache::MatchingMarks> QueryConditionCache::read(const UUID & table_id, const String & part_name, size_t predicate_hash)
+std::optional<QueryConditionCache::MatchingMarks> QueryConditionCache::read(const UUID & table_id, const String & part_name, size_t condition_hash)
 {
-    Key key = {table_id, part_name, predicate_hash};
+    Key key = {table_id, part_name, condition_hash};
 
     if (auto entry = cache.get(key))
     {
         ProfileEvents::increment(ProfileEvents::QueryConditionCacheHits);
 
-        std::lock_guard lock(entry->mutex);
+        std::shared_lock lock(entry->mutex);
 
         LOG_DEBUG(
             logger,
-            "Read entry for table_uuid: {}, part: {}, predicate_hash: {}, ranges: {}",
+            "Read entry for table_uuid: {}, part: {}, condition_hash: {}, ranges: {}",
             table_id,
             part_name,
-            predicate_hash,
+            condition_hash,
             toString(entry->matching_marks));
 
         return {entry->matching_marks};
@@ -75,16 +78,21 @@ std::optional<QueryConditionCache::MatchingMarks> QueryConditionCache::read(cons
 
         LOG_DEBUG(
             logger,
-            "Could not find entry for table_uuid: {}, part: {}, predicate_hash: {}",
+            "Could not find entry for table_uuid: {}, part: {}, condition_hash: {}",
             table_id,
             part_name,
-            predicate_hash);
+            condition_hash);
 
         return {};
     }
 
 }
 
+std::vector<QueryConditionCache::Cache::KeyMapped> QueryConditionCache::dump() const
+{
+    return cache.dump();
+}
+
 void QueryConditionCache::clear()
 {
     cache.clear();
@@ -99,7 +107,7 @@ bool QueryConditionCache::Key::operator==(const Key & other) const
 {
     return table_id == other.table_id
         && part_name == other.part_name
-        && predicate_hash == other.predicate_hash;
+        && condition_hash == other.condition_hash;
 }
 
 QueryConditionCache::Entry::Entry(size_t mark_count)
@@ -112,7 +120,7 @@ size_t QueryConditionCache::KeyHasher::operator()(const Key & key) const
     SipHash hash;
     hash.update(key.table_id);
     hash.update(key.part_name);
-    hash.update(key.predicate_hash);
+    hash.update(key.condition_hash);
     return hash.get64();
 }
 

src/Interpreters/Cache/QueryConditionCache.h

@@ -2,6 +2,7 @@
 
 #include <Common/CacheBase.h>
 #include <Storages/MergeTree/MarkRange.h>
+#include <shared_mutex>
 
 namespace DB
 {
@@ -19,35 +20,21 @@ class QueryConditionCache
     /// True means at least one row in the mark matches the predicate. We need to read such marks.
     using MatchingMarks = std::vector<bool>;
 
-    QueryConditionCache(const String & cache_policy, size_t max_size_in_bytes, double size_ratio);
-
-    /// Add an entry to the cache. The passed marks represent ranges of the column with matches of the predicate.
-    void write(
-        const UUID & table_id, const String & part_name, size_t predicate_hash,
-        const MarkRanges & mark_ranges, size_t marks_count, bool has_final_mark);
-
-    /// Check the cache if it contains an entry for the given table + part id and predicate hash.
-    std::optional<MatchingMarks> read(const UUID & table_id, const String & part_name, size_t predicate_hash);
-
-    void clear();
-
-    void setMaxSizeInBytes(size_t max_size_in_bytes);
-
 private:
     /// Key + entry represent a mark range result.
     struct Key
     {
         const UUID table_id;
         const String part_name;
-        const size_t predicate_hash;
+        const size_t condition_hash;
 
         bool operator==(const Key & other) const;
     };
 
     struct Entry
     {
         MatchingMarks matching_marks;
-        std::mutex mutex; /// (*)
+        std::shared_mutex mutex; /// (*)
 
         explicit Entry(size_t mark_count);
     };
@@ -67,9 +54,28 @@ class QueryConditionCache
         size_t operator()(const Entry & entry) const;
     };
 
+public:
     using Cache = CacheBase<Key, Entry, KeyHasher, QueryConditionCacheEntryWeight>;
-    Cache cache;
 
+    QueryConditionCache(const String & cache_policy, size_t max_size_in_bytes, double size_ratio);
+
+    /// Add an entry to the cache. The passed marks represent ranges of the column with matches of the predicate.
+    void write(
+        const UUID & table_id, const String & part_name, size_t condition_hash,
+        const MarkRanges & mark_ranges, size_t marks_count, bool has_final_mark);
+
+    /// Check the cache if it contains an entry for the given table + part id and predicate hash.
+    std::optional<MatchingMarks> read(const UUID & table_id, const String & part_name, size_t condition_hash);
+
+    /// For debugging and system tables
+    std::vector<QueryConditionCache::Cache::KeyMapped> dump() const;
+
+    void clear();
+
+    void setMaxSizeInBytes(size_t max_size_in_bytes);
+
+private:
+    Cache cache;
     LoggerPtr logger = getLogger("QueryConditionCache");
 };
 

src/Processors/Chunk.h

@@ -2,7 +2,7 @@
 
 #include <Columns/IColumn_fwd.h>
 #include <Common/CollectionOfDerived.h>
-#include <Columns/IColumn.h>
+#include <Core/Types_fwd.h>
 #include <Storages/MergeTree/MarkRange.h>
 
 #include <memory>
@@ -158,21 +158,23 @@ class AsyncInsertInfo : public ChunkInfoCloneable<AsyncInsertInfo>
 
 using AsyncInsertInfoPtr = std::shared_ptr<AsyncInsertInfo>;
 
-class IMergeTreeDataPart;
-
-/// The query condition cache needs to know the mark ranges of which part the chunk data comes from.
+/// Lineage information: from which table, part and mark range does the chunk come from?
+/// This information is needed by the query condition cache.
 class MarkRangesInfo : public ChunkInfoCloneable<MarkRangesInfo>
 {
 public:
-    MarkRangesInfo(std::shared_ptr<const IMergeTreeDataPart> data_part_, MarkRanges mark_ranges_)
-        : data_part(data_part_)
+    MarkRangesInfo(UUID table_uuid_, const String & part_name_, size_t marks_count_, bool has_final_mark_, MarkRanges mark_ranges_)
+        : table_uuid(table_uuid_)
+        , part_name(part_name_)
+        , marks_count(marks_count_)
+        , has_final_mark(has_final_mark_)
         , mark_ranges(std::move(mark_ranges_))
     {}
 
-    std::shared_ptr<const IMergeTreeDataPart> getDataPart() const { return data_part; }
-    const MarkRanges & getMarkRanges() const { return mark_ranges; }
-private:
-    std::shared_ptr<const IMergeTreeDataPart> data_part;
+    UUID table_uuid;
+    String part_name;
+    size_t marks_count;
+    bool has_final_mark;
     MarkRanges mark_ranges;
 };
 

src/Processors/QueryPlan/FilterStep.cpp

@@ -173,11 +173,7 @@ void FilterStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQ
     pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type)
     {
         bool on_totals = stream_type == QueryPipelineBuilder::StreamType::Totals;
-        if (condition_hash)
-            return std::make_shared<FilterTransform>(header, expression, filter_column_name, remove_filter_column,
-                on_totals, nullptr, condition_hash);
-
-        return std::make_shared<FilterTransform>(header, expression, filter_column_name, remove_filter_column, on_totals);
+        return std::make_shared<FilterTransform>(header, expression, filter_column_name, remove_filter_column, on_totals, nullptr, condition_hash);
     });
 
     if (!blocksHaveEqualStructure(pipeline.getHeader(), *output_header))
@@ -252,7 +248,7 @@ void FilterStep::updateOutputHeader()
         return;
 }
 
-void FilterStep::setQueryConditionKey(size_t condition_hash_)
+void FilterStep::setQueryConditionHash(size_t condition_hash_)
 {
     condition_hash = condition_hash_;
 }

src/Processors/QueryPlan/FilterStep.h

@@ -1,7 +1,6 @@
 #pragma once
 #include <Processors/QueryPlan/ITransformingStep.h>
 #include <Interpreters/ActionsDAG.h>
-#include <Interpreters/Cache/QueryConditionCache.h>
 
 namespace DB
 {
@@ -26,7 +25,8 @@ class FilterStep : public ITransformingStep
     ActionsDAG & getExpression() { return actions_dag; }
     const String & getFilterColumnName() const { return filter_column_name; }
     bool removesFilterColumn() const { return remove_filter_column; }
-    void setQueryConditionKey(size_t condition_hash_);
+
+    void setQueryConditionHash(size_t condition_hash_);
 
     static bool canUseType(const DataTypePtr & type);
 

src/Processors/QueryPlan/Optimizations/Optimizations.h

@@ -128,7 +128,7 @@ bool optimizeJoinLegacy(QueryPlan::Node & node, QueryPlan::Nodes &, const QueryP
 bool optimizeJoinLogical(QueryPlan::Node & node, QueryPlan::Nodes &, const QueryPlanOptimizationSettings &);
 bool convertLogicalJoinToPhysical(QueryPlan::Node & node, QueryPlan::Nodes &, const QueryPlanOptimizationSettings & optimization_settings);
 void optimizeDistinctInOrder(QueryPlan::Node & node, QueryPlan::Nodes &);
-void tryUpdateQueryConditionCache(const QueryPlanOptimizationSettings & optimization_settings, const Stack & stack);
+void updateQueryConditionCache(const Stack & stack, const QueryPlanOptimizationSettings & optimization_settings);
 
 // Should be called once the query plan tree structure is finalized, i.e. no nodes addition, deletion or pushing down should happen after that call.
 // Since those hashes are used for join optimization, the calculation performed before join optimization.

src/Processors/QueryPlan/Optimizations/QueryPlanOptimizationSettings.cpp

@@ -8,6 +8,7 @@ namespace DB
 namespace Setting
 {
     extern const SettingsBool allow_aggregate_partitions_independently;
+    extern const SettingsBool allow_experimental_analyzer;
     extern const SettingsBool force_optimize_projection;
     extern const SettingsBool optimize_aggregation_in_order;
     extern const SettingsBool optimize_distinct_in_order;
@@ -30,16 +31,15 @@ namespace Setting
     extern const SettingsBool query_plan_remove_redundant_distinct;
     extern const SettingsBool query_plan_remove_redundant_sorting;
     extern const SettingsBool query_plan_reuse_storage_ordering_for_window_functions;
-    extern const SettingsBoolAuto query_plan_join_swap_table;
     extern const SettingsBool query_plan_split_filter;
     extern const SettingsBool query_plan_try_use_vector_search;
+    extern const SettingsBool use_query_condition_cache;
+    extern const SettingsBoolAuto query_plan_join_swap_table;
+    extern const SettingsMaxThreads max_threads;
+    extern const SettingsSeconds lock_acquire_timeout;
     extern const SettingsString force_optimize_projection_name;
     extern const SettingsUInt64 max_limit_for_ann_queries;
-    extern const SettingsSeconds lock_acquire_timeout;
-    extern const SettingsMaxThreads max_threads;
     extern const SettingsUInt64 query_plan_max_optimizations_to_apply;
-    extern const SettingsBool use_query_condition_cache;
-    extern const SettingsBool allow_experimental_analyzer;
 }
 
 namespace ServerSetting
@@ -81,6 +81,7 @@ QueryPlanOptimizationSettings::QueryPlanOptimizationSettings(
     optimize_sorting_by_input_stream_properties = from[Setting::query_plan_enable_optimizations] && from[Setting::optimize_sorting_by_input_stream_properties];
     aggregation_in_order = from[Setting::query_plan_enable_optimizations] && from[Setting::optimize_aggregation_in_order] && from[Setting::query_plan_aggregation_in_order];
     optimize_projection = from[Setting::optimize_use_projections];
+    use_query_condition_cache = from[Setting::use_query_condition_cache] && from[Setting::allow_experimental_analyzer];
 
     optimize_use_implicit_projections = optimize_projection && from[Setting::optimize_use_implicit_projections];
     force_use_projection = optimize_projection && from[Setting::force_optimize_projection];
@@ -90,7 +91,6 @@ QueryPlanOptimizationSettings::QueryPlanOptimizationSettings(
 
     /// These settings comes from EXPLAIN settings not query settings and outside of the scope of this class
     keep_logical_steps = false;
-    use_query_condition_cache = from[Setting::use_query_condition_cache] && from[Setting::allow_experimental_analyzer];
     is_explain = false;
 
     max_entries_for_hash_table_stats = max_entries_for_hash_table_stats_;

src/Processors/QueryPlan/Optimizations/QueryPlanOptimizationSettings.h

@@ -61,6 +61,7 @@ struct QueryPlanOptimizationSettings
     bool optimize_sorting_by_input_stream_properties;
     bool aggregation_in_order;
     bool optimize_projection;
+    bool use_query_condition_cache = false;
 
     /// --- Third-pass optimizations (Processors/QueryPlan/QueryPlan.cpp)
     bool build_sets = true; /// this one doesn't have a corresponding setting
@@ -91,8 +92,6 @@ struct QueryPlanOptimizationSettings
 
     bool keep_logical_steps;
 
-    /// If query condition cache is enabled, the query condition cache needs to be updated in the WHERE stage.
-    bool use_query_condition_cache = false;
     bool is_explain;
 };
 

src/Processors/QueryPlan/Optimizations/optimizeTree.cpp

@@ -124,7 +124,7 @@ void optimizeTreeSecondPass(const QueryPlanOptimizationSettings & optimization_s
     {
         optimizePrimaryKeyConditionAndLimit(stack);
 
-        tryUpdateQueryConditionCache(optimization_settings, stack);
+        updateQueryConditionCache(stack, optimization_settings);
 
         /// NOTE: optimizePrewhere can modify the stack.
         /// Prewhere optimization relies on PK optimization (getConditionSelectivityEstimatorByPredicate)