From 7e158ab1eaf5404fcb87b0c6d36ef6b5bb571326 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Thu, 30 Apr 2026 14:59:57 +0800 Subject: [PATCH] Backfill Unobserved Vectors Signed-off-by: peterxcli --- CMakeLists.txt | 2 + src/include/logical_cache_recorder.hpp | 41 +++ src/include/physical_cache_recorder.hpp | 60 ++++ .../query_condition_cache_functions.hpp | 3 + .../query_condition_cache_optimizer.hpp | 42 +-- src/include/query_condition_cache_state.hpp | 46 ++- src/logical_cache_recorder.cpp | 129 ++++++++ src/physical_cache_recorder.cpp | 214 +++++++++++++ src/query_condition_cache_extension.cpp | 3 + src/query_condition_cache_functions.cpp | 59 +++- src/query_condition_cache_optimizer.cpp | 295 +++++++++++++----- src/query_condition_cache_state.cpp | 232 ++++++++++++-- test/sql/condition_cache_auto.test | 2 +- test/sql/condition_cache_build.test | 2 +- test/sql/condition_cache_explain.test | 51 +++ test/sql/condition_cache_incremental.test | 94 ++++++ test/sql/condition_cache_info.test | 2 +- test/unittest/CMakeLists.txt | 1 + test/unittest/test_bitvector.cpp | 78 +++++ test/unittest/test_build_cache_entry.cpp | 13 + test/unittest/test_filter.cpp | 72 +++++ .../unittest/test_physical_cache_recorder.cpp | 127 ++++++++ 22 files changed, 1429 insertions(+), 139 deletions(-) create mode 100644 src/include/logical_cache_recorder.hpp create mode 100644 src/include/physical_cache_recorder.hpp create mode 100644 src/logical_cache_recorder.cpp create mode 100644 src/physical_cache_recorder.cpp create mode 100644 test/sql/condition_cache_explain.test create mode 100644 test/sql/condition_cache_incremental.test create mode 100644 test/unittest/test_physical_cache_recorder.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index bfabf4f..7749aab 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -21,7 +21,9 @@ include_directories(src/include) set(EXTENSION_SOURCES src/cache_invalidation_optimizer.cpp + src/logical_cache_recorder.cpp src/logical_cache_invalidator.cpp + src/physical_cache_recorder.cpp src/physical_cache_invalidator.cpp src/predicate_key_utils.cpp src/query_condition_cache_extension.cpp diff --git a/src/include/logical_cache_recorder.hpp b/src/include/logical_cache_recorder.hpp new file mode 100644 index 0000000..ddfc4ad --- /dev/null +++ b/src/include/logical_cache_recorder.hpp @@ -0,0 +1,41 @@ +#pragma once + +#include "duckdb/planner/operator/logical_extension_operator.hpp" +#include "physical_cache_recorder.hpp" + +namespace duckdb { + +struct LogicalCacheRecorder : public LogicalExtensionOperator { + idx_t table_oid; + string table_catalog; + string table_schema; + string table_name; + string canonical_key; + idx_t rowid_column_index; + shared_ptr cache_entry; + shared_ptr metadata_entry; + + LogicalCacheRecorder(idx_t table_oid_p, string canonical_key_p, unique_ptr bound_predicate_p, + idx_t rowid_column_index_p, string table_catalog_p = string(), + string table_schema_p = string(), string table_name_p = string(), + unique_ptr backfill_predicate_p = nullptr, + shared_ptr cache_entry_p = nullptr, + shared_ptr metadata_entry_p = nullptr); + + PhysicalOperator &CreatePlan(ClientContext &context, PhysicalPlanGenerator &planner) override; + vector GetColumnBindings() override; + string GetExtensionName() const override; + void Serialize(Serializer &serializer) const override; + +protected: + void ResolveTypes() override; +}; + +class CacheRecorderOperatorExtension : public OperatorExtension { +public: + CacheRecorderOperatorExtension(); + string GetName() override; + unique_ptr Deserialize(Deserializer &deserializer) override; +}; + +} // namespace duckdb diff --git a/src/include/physical_cache_recorder.hpp b/src/include/physical_cache_recorder.hpp new file mode 100644 index 0000000..5ec089a --- /dev/null +++ b/src/include/physical_cache_recorder.hpp @@ -0,0 +1,60 @@ +#pragma once + +#include "concurrency/annotated_mutex.hpp" +#include "concurrency/thread_annotation.hpp" +#include "query_condition_cache_state.hpp" + +#include "duckdb/execution/expression_executor.hpp" +#include "duckdb/execution/physical_operator.hpp" +#include "duckdb/planner/expression.hpp" + +namespace duckdb { + +struct CacheRecorderLocalState : public OperatorState { + CacheRecorderLocalState(ClientContext &context, const Expression &bound_predicate); + + shared_ptr local_entry; + ExpressionExecutor expr_executor; +}; + +struct CacheRecorderGlobalState : public GlobalOperatorState { + concurrency::mutex lock; + vector> task_local_entries DUCKDB_GUARDED_BY(lock); +}; + +class PhysicalCacheRecorder : public PhysicalOperator { +public: + PhysicalCacheRecorder(PhysicalPlan &physical_plan, idx_t table_oid_p, string canonical_key_p, + unique_ptr bound_predicate_p, idx_t rowid_column_index_p, string table_catalog_p, + string table_schema_p, string table_name_p, unique_ptr backfill_predicate_p, + shared_ptr cache_entry_p, + shared_ptr metadata_entry_p, vector types, + idx_t estimated_cardinality); + + idx_t table_oid; + string canonical_key; + string table_catalog; + string table_schema; + string table_name; + unique_ptr bound_predicate; + unique_ptr backfill_predicate; + idx_t rowid_column_index; + shared_ptr cache_entry; + shared_ptr metadata_entry; + + unique_ptr GetGlobalOperatorState(ClientContext &context) const override; + unique_ptr GetOperatorState(ExecutionContext &context) const override; + OperatorResultType Execute(ExecutionContext &context, DataChunk &input, DataChunk &chunk, + GlobalOperatorState &gstate, OperatorState &state) const override; + OperatorFinalResultType OperatorFinalize(Pipeline &pipeline, Event &event, ClientContext &context, + OperatorFinalizeInput &input) const override; + bool RequiresOperatorFinalize() const override; + bool ParallelOperator() const override; + string GetName() const override; + InsertionOrderPreservingMap ParamsToString() const override; + + static void RecordChunkObservation(ConditionCacheEntry &local_entry, idx_t rg_idx, idx_t vec_idx, + bool has_qualifying); +}; + +} // namespace duckdb diff --git a/src/include/query_condition_cache_functions.hpp b/src/include/query_condition_cache_functions.hpp index 850e253..f3e9f83 100644 --- a/src/include/query_condition_cache_functions.hpp +++ b/src/include/query_condition_cache_functions.hpp @@ -15,6 +15,9 @@ class Expression; // Modifies bound_expr in-place (remaps column indices to scan positions). shared_ptr BuildCacheEntry(ClientContext &context, DuckTableEntry &table_entry, Expression &bound_expr); +shared_ptr BuildCacheEntryForRanges(ClientContext &context, DuckTableEntry &table_entry, + Expression &bound_expr, + const vector &ranges); TableFunction ConditionCacheBuildFunction(); TableFunction ConditionCacheInfoFunction(); diff --git a/src/include/query_condition_cache_optimizer.hpp b/src/include/query_condition_cache_optimizer.hpp index 221646a..767c52d 100644 --- a/src/include/query_condition_cache_optimizer.hpp +++ b/src/include/query_condition_cache_optimizer.hpp @@ -4,23 +4,33 @@ #include "duckdb/main/client_context_state.hpp" #include "duckdb/optimizer/optimizer_extension.hpp" +#include "duckdb/planner/expression.hpp" namespace duckdb { -class DuckTableEntry; class LogicalGet; -// Query-scoped state for passing cache entries between pre-optimize and post-optimize phases. -// Stored in ClientContext::registered_state; automatically cleared on QueryEnd. +struct RecorderInjectionInfo { + idx_t table_oid; + string table_catalog; + string table_schema; + string table_name; + string canonical_key; + unique_ptr predicate; + unique_ptr backfill_predicate; + shared_ptr entry; + shared_ptr metadata_entry; +}; + struct CacheOptimizerQueryState : public ClientContextState { static constexpr const char *NAME = "qcc_optimizer_state"; - // Maps table_index -> cache entry for tables matched during pre-optimize. - // Consumed by post-optimize to inject cache filters. unordered_map> cache_apply_pending; + unordered_map cache_recorder_pending; void QueryEnd(ClientContext &context, optional_ptr error) override { cache_apply_pending.clear(); + cache_recorder_pending.clear(); } }; @@ -28,31 +38,27 @@ class QueryConditionCacheOptimizer : public OptimizerExtension { public: QueryConditionCacheOptimizer(); - // Pre-optimize: compute canonical predicate keys before FilterPushdown splits the WHERE clause. - // On cache miss, builds cache inline so the first query benefits immediately. static void PreOptimizeFunction(OptimizerExtensionInput &input, unique_ptr &plan); - - // Post-optimize: inject cache filters into LogicalGet nodes that were matched pre-optimize. static void OptimizeFunction(OptimizerExtensionInput &input, unique_ptr &plan); private: static bool IsSettingEnabled(ClientContext &context); - // Walk plan before FilterPushdown: find LogicalFilter -> LogicalGet, compute key, lookup/build cache static void PreOptimizeWalk(ClientContext &context, unique_ptr &plan, bool inside_dml, - CacheOptimizerQueryState &state); - - // Build cache entry for a predicate on a table - static shared_ptr - BuildCacheForPredicate(ClientContext &context, const vector> &expressions, LogicalGet &get); - - // Walk plan after built-in optimization and inject cache filters into matching table scans. + bool inside_truncating, CacheOptimizerQueryState &state); static void PostOptimizeWalk(ClientContext &context, unique_ptr &plan, CacheOptimizerQueryState &state); - // Inject a rowid-backed cache filter into a LogicalGet while preserving its visible output. static void InjectCacheFilter(ClientContext &context, LogicalGet &get, const shared_ptr &entry); + static void InjectCacheRecorder(unique_ptr &plan, idx_t table_oid, string canonical_key, + string table_catalog, string table_schema, string table_name, + unique_ptr predicate, unique_ptr backfill_predicate, + const shared_ptr &entry, + const shared_ptr &metadata_entry); + static shared_ptr GetPrunedRowGroupsFromTableFilters(ClientContext &context, + const LogicalGet &get); + static idx_t EnsureRowIdChunkIndex(LogicalGet &get); }; } // namespace duckdb diff --git a/src/include/query_condition_cache_state.hpp b/src/include/query_condition_cache_state.hpp index 5be7bba..e7d248f 100644 --- a/src/include/query_condition_cache_state.hpp +++ b/src/include/query_condition_cache_state.hpp @@ -4,7 +4,8 @@ #include "concurrency/annotated_mutex.hpp" #include "concurrency/thread_annotation.hpp" -#include "duckdb/common/array.hpp" +#include + #include "duckdb/common/types/hash.hpp" #include "duckdb/common/unordered_map.hpp" #include "duckdb/common/unordered_set.hpp" @@ -14,25 +15,28 @@ namespace duckdb { // Derived from DuckDB's compile-time configurable constants inline constexpr idx_t VECTORS_PER_ROW_GROUP = DEFAULT_ROW_GROUP_SIZE / STANDARD_VECTOR_SIZE; -inline constexpr idx_t BITVECTOR_ARRAY_SIZE = (VECTORS_PER_ROW_GROUP + 63) / 64; static_assert(DEFAULT_ROW_GROUP_SIZE % STANDARD_VECTOR_SIZE == 0, "DEFAULT_ROW_GROUP_SIZE must be divisible by STANDARD_VECTOR_SIZE"); -// Per row-group bitvector: bit[i] = 1 means vector i has at least one qualifying row, -// for 0 <= i < VECTORS_PER_ROW_GROUP. -// Backed by array to support configurable row group / vector sizes. +// Per row-group pair of bitsets: +// matching_vectors[i] = 1 iff vec i has at least one qualifying row. +// observed[i] = 1 iff vec i has been observed by an exact scan path. struct RowGroupFilter { - array matching_vectors = {}; + std::bitset matching_vectors; + std::bitset observed; RowGroupFilter() = default; - // Construct from vector indices that contain at least one qualifying row. - // Each index must be in [0, VECTORS_PER_ROW_GROUP). Duplicates are allowed. + // Leaves observed all-zero; call SetObserved explicitly when a vec was scanned. explicit RowGroupFilter(const vector &qualifying_vectors); void SetVector(idx_t vector_index); bool VectorHasRows(idx_t vector_index) const; bool IsEmpty() const; + void SetObserved(idx_t vector_index); + void MarkFullyObserved(); + bool IsObserved(idx_t vector_index) const; + bool IsFullyObserved() const; void MergeFrom(const RowGroupFilter &other); }; @@ -55,10 +59,15 @@ struct CacheKeyHashFunction { struct CacheEntryStats { idx_t qualifying_vectors; idx_t total_vectors; - idx_t qualifying_row_groups; + idx_t cached_row_groups; idx_t total_row_groups; }; +struct CacheObservationRange { + idx_t start_row; + idx_t count; +}; + // A single cache entry: the bitvectors for one (table, predicate) combination. struct ConditionCacheEntry : public ObjectCacheEntry { static string ObjectType() { @@ -72,7 +81,7 @@ struct ConditionCacheEntry : public ObjectCacheEntry { // Return estimated memory usage for LRU eviction optional_idx GetEstimatedCacheMemory() const override; - // Compute statistics about qualifying vectors and row groups + // Compute statistics about qualifying vectors and cached row groups CacheEntryStats ComputeStats(idx_t total_rows) const; // --- Thread-safe API (each method acquires `lock` internally) --- @@ -83,20 +92,30 @@ struct ConditionCacheEntry : public ObjectCacheEntry { void SetQualifyingVector(idx_t rg_idx, idx_t vec_idx); // Merge another entry's row-group filters into this entry (e.g. after parallel build). void MergeFrom(const ConditionCacheEntry &other); + // Used by full-table builds to declare every row group fully observed. + void MarkAllRowGroupsFullyObserved(); + void MarkRowGroupFullyObserved(idx_t rg_idx); + // Mark that a specific vector was observed by an exact scan path. + void SetObservedVector(idx_t rg_idx, idx_t vec_idx); - // Row group absent from cache, or vector has qualifying rows -> predicate may pass rows (matches scan semantics). + // Row group absent from cache, or vector not observed yet, or vector has qualifying rows -> pass rows through. bool VectorPassesFilter(idx_t rg_idx, idx_t vec_idx) const; - // True iff every row group in [min_rg, max_rg] is present in the cache and has an empty filter. + // True iff every row group in [min_rg, max_rg] is present in the cache, empty, and fully observed. bool StatisticsRangeIsAllEmptyCached(idx_t min_rg, idx_t max_rg) const; + // True iff some row group/vector covering [0, total_rows) is absent or not fully observed. + bool NeedsObservation(idx_t total_rows) const; + vector GetUnobservedVectorRanges(idx_t total_rows) const; idx_t RowGroupCount() const; bool HasRowGroup(idx_t rg_idx) const; + idx_t GetObservedVectorCount(idx_t rg_idx) const; bool RowGroupVectorHasQualifyingRows(idx_t rg_idx, idx_t vec_idx) const; // True iff `rg_idx` is cached and its filter is empty (no qualifying vectors). bool RowGroupIsCompletelyEmpty(idx_t rg_idx) const; // Erase row group keys; returns (number of keys removed, whether the map is now empty). pair EraseRowGroups(const unordered_set &row_group_indices); + pair EraseRowGroupsStartingAt(idx_t first_row_group); private: mutable concurrency::mutex lock; @@ -159,6 +178,8 @@ class ConditionCacheStore : public ObjectCacheEntry { // Remove specific row groups from all entries for a table. Returns count of row groups removed. idx_t RemoveRowGroupsForTable(ClientContext &context, idx_t table_oid, const unordered_set &row_group_indices); + idx_t RemoveRowGroupsStartingAtForTable(ClientContext &context, idx_t table_oid, idx_t first_row_group); + idx_t RemoveRowGroupsStartingAtForTable(DatabaseInstance &db, idx_t table_oid, idx_t first_row_group); // Check if any entries exist for a given table OID bool HasEntriesForTable(ClientContext &context, idx_t table_oid); @@ -168,6 +189,7 @@ class ConditionCacheStore : public ObjectCacheEntry { // Get or create the store from a client context static shared_ptr GetOrCreate(ClientContext &context); + static shared_ptr GetOrCreate(DatabaseInstance &db); private: concurrency::mutex lock; diff --git a/src/logical_cache_recorder.cpp b/src/logical_cache_recorder.cpp new file mode 100644 index 0000000..1c69ed4 --- /dev/null +++ b/src/logical_cache_recorder.cpp @@ -0,0 +1,129 @@ +#include "logical_cache_recorder.hpp" + +#include "duckdb/common/exception.hpp" +#include "duckdb/common/serializer/deserializer.hpp" +#include "duckdb/common/serializer/serializer.hpp" +#include "duckdb/execution/physical_plan_generator.hpp" +#include "duckdb/planner/expression/bound_columnref_expression.hpp" +#include "duckdb/planner/expression/bound_constant_expression.hpp" +#include "duckdb/planner/expression/bound_reference_expression.hpp" +#include "duckdb/planner/expression_iterator.hpp" + +namespace duckdb { + +namespace { + +void ConvertColumnRefsToChunkRefs(unique_ptr &expr, const vector &bindings) { + if (expr->GetExpressionClass() == ExpressionClass::BOUND_COLUMN_REF) { + auto &colref = expr->Cast(); + for (idx_t i = 0; i < bindings.size(); ++i) { + if (bindings[i] == colref.binding) { + expr = make_uniq(colref.alias, colref.return_type, i); + return; + } + } + throw InternalException("Failed to bind cache recorder predicate column reference"); + } + ExpressionIterator::EnumerateChildren( + *expr, [&](unique_ptr &child) { ConvertColumnRefsToChunkRefs(child, bindings); }); +} + +} // namespace + +LogicalCacheRecorder::LogicalCacheRecorder(idx_t table_oid_p, string canonical_key_p, + unique_ptr bound_predicate_p, idx_t rowid_column_index_p, + string table_catalog_p, string table_schema_p, string table_name_p, + unique_ptr backfill_predicate_p, + shared_ptr cache_entry_p, + shared_ptr metadata_entry_p) + : table_oid(table_oid_p), table_catalog(std::move(table_catalog_p)), table_schema(std::move(table_schema_p)), + table_name(std::move(table_name_p)), canonical_key(std::move(canonical_key_p)), + rowid_column_index(rowid_column_index_p), cache_entry(std::move(cache_entry_p)), + metadata_entry(std::move(metadata_entry_p)) { + expressions.push_back(std::move(bound_predicate_p)); + if (backfill_predicate_p) { + expressions.push_back(std::move(backfill_predicate_p)); + } +} + +PhysicalOperator &LogicalCacheRecorder::CreatePlan(ClientContext &context, PhysicalPlanGenerator &planner) { + auto &child_plan = planner.CreatePlan(*children[0]); + auto bound_predicate = std::move(expressions[0]); + ConvertColumnRefsToChunkRefs(bound_predicate, children[0]->GetColumnBindings()); + unique_ptr backfill_predicate; + if (expressions.size() > 1) { + backfill_predicate = std::move(expressions[1]); + } + auto &op = + planner.Make(table_oid, canonical_key, std::move(bound_predicate), rowid_column_index, + table_catalog, table_schema, table_name, std::move(backfill_predicate), + cache_entry, metadata_entry, child_plan.types, estimated_cardinality); + op.children.push_back(child_plan); + return op; +} + +vector LogicalCacheRecorder::GetColumnBindings() { + return children[0]->GetColumnBindings(); +} + +void LogicalCacheRecorder::ResolveTypes() { + types = children[0]->types; +} + +string LogicalCacheRecorder::GetExtensionName() const { + return "query_condition_cache_recorder"; +} + +void LogicalCacheRecorder::Serialize(Serializer &serializer) const { + LogicalExtensionOperator::Serialize(serializer); + serializer.WriteProperty(400, "table_oid", table_oid); + serializer.WriteProperty(401, "canonical_key", canonical_key); + serializer.WriteProperty(402, "rowid_column_index", rowid_column_index); + serializer.WritePropertyWithDefault(403, "expressions", expressions); + serializer.WritePropertyWithDefault(404, "table_catalog", table_catalog); + serializer.WritePropertyWithDefault(405, "table_schema", table_schema); + serializer.WritePropertyWithDefault(406, "table_name", table_name); +} + +namespace { + +BoundStatement CacheRecorderBind(ClientContext &context, Binder &binder, OperatorExtensionInfo *info, + SQLStatement &statement) { + return BoundStatement(); +} + +} // namespace + +CacheRecorderOperatorExtension::CacheRecorderOperatorExtension() { + Bind = CacheRecorderBind; +} + +string CacheRecorderOperatorExtension::GetName() { + return "query_condition_cache_recorder"; +} + +unique_ptr CacheRecorderOperatorExtension::Deserialize(Deserializer &deserializer) { + auto oid = deserializer.ReadProperty(400, "table_oid"); + auto key = deserializer.ReadProperty(401, "canonical_key"); + auto rowid_col = deserializer.ReadProperty(402, "rowid_column_index"); + auto exprs = deserializer.ReadPropertyWithDefault>>(403, "expressions"); + auto table_catalog = deserializer.ReadPropertyWithDefault(404, "table_catalog"); + auto table_schema = deserializer.ReadPropertyWithDefault(405, "table_schema"); + auto table_name = deserializer.ReadPropertyWithDefault(406, "table_name"); + + unique_ptr bound_predicate; + if (!exprs.empty()) { + bound_predicate = std::move(exprs[0]); + } else { + bound_predicate = make_uniq(Value::BOOLEAN(true)); + } + unique_ptr backfill_predicate; + if (exprs.size() > 1) { + backfill_predicate = std::move(exprs[1]); + } + return make_uniq(oid, std::move(key), std::move(bound_predicate), rowid_col, + std::move(table_catalog), std::move(table_schema), std::move(table_name), + std::move(backfill_predicate)); +} + +} // namespace duckdb diff --git a/src/physical_cache_recorder.cpp b/src/physical_cache_recorder.cpp new file mode 100644 index 0000000..2d406cb --- /dev/null +++ b/src/physical_cache_recorder.cpp @@ -0,0 +1,214 @@ +#include "physical_cache_recorder.hpp" + +#include "concurrency/annotated_lock.hpp" +#include "query_condition_cache_functions.hpp" + +#include "duckdb/catalog/catalog.hpp" +#include "duckdb/catalog/catalog_entry/duck_table_entry.hpp" +#include "duckdb/common/assert.hpp" +#include "duckdb/common/numeric_utils.hpp" +#include "duckdb/common/unordered_set.hpp" +#include "duckdb/common/types/selection_vector.hpp" +#include "duckdb/main/client_context.hpp" + +namespace duckdb { + +CacheRecorderLocalState::CacheRecorderLocalState(ClientContext &context, const Expression &bound_predicate) + : expr_executor(context, bound_predicate) { +} + +PhysicalCacheRecorder::PhysicalCacheRecorder(PhysicalPlan &physical_plan, idx_t table_oid_p, string canonical_key_p, + unique_ptr bound_predicate_p, idx_t rowid_column_index_p, + string table_catalog_p, string table_schema_p, string table_name_p, + unique_ptr backfill_predicate_p, + shared_ptr cache_entry_p, + shared_ptr metadata_entry_p, + vector types, idx_t estimated_cardinality) + : PhysicalOperator(physical_plan, PhysicalOperatorType::EXTENSION, std::move(types), estimated_cardinality), + table_oid(table_oid_p), canonical_key(std::move(canonical_key_p)), table_catalog(std::move(table_catalog_p)), + table_schema(std::move(table_schema_p)), table_name(std::move(table_name_p)), + bound_predicate(std::move(bound_predicate_p)), backfill_predicate(std::move(backfill_predicate_p)), + rowid_column_index(rowid_column_index_p), cache_entry(std::move(cache_entry_p)), + metadata_entry(std::move(metadata_entry_p)) { +} + +unique_ptr PhysicalCacheRecorder::GetGlobalOperatorState(ClientContext &context) const { + return make_uniq(); +} + +unique_ptr PhysicalCacheRecorder::GetOperatorState(ExecutionContext &context) const { + return make_uniq(context.client, *bound_predicate); +} + +namespace { + +void RegisterLocalIfNeeded(CacheRecorderLocalState &local_state, CacheRecorderGlobalState &global_state) { + if (local_state.local_entry) { + return; + } + auto entry = make_shared_ptr(); + { + concurrency::lock_guard guard(global_state.lock); + global_state.task_local_entries.push_back(entry); + } + local_state.local_entry = std::move(entry); +} + +bool TryEncodeVectorKey(row_t row_id, idx_t &key) { + if (row_id < 0 || row_id >= MAX_ROW_ID) { + return false; + } + auto unsigned_row_id = NumericCast(row_id); + auto rg_idx = unsigned_row_id / DEFAULT_ROW_GROUP_SIZE; + auto vec_idx = (unsigned_row_id % DEFAULT_ROW_GROUP_SIZE) / STANDARD_VECTOR_SIZE; + key = rg_idx * VECTORS_PER_ROW_GROUP + vec_idx; + return true; +} + +} // namespace + +OperatorResultType PhysicalCacheRecorder::Execute(ExecutionContext &context, DataChunk &input, DataChunk &chunk, + GlobalOperatorState &gstate, OperatorState &state) const { + D_ASSERT(rowid_column_index < input.ColumnCount()); + + chunk.Reference(input); + + auto &local_state = state.Cast(); + auto &global_state = gstate.Cast(); + RegisterLocalIfNeeded(local_state, global_state); + + if (input.size() == 0) { + return OperatorResultType::NEED_MORE_INPUT; + } + + auto &rowid_vec = input.data[rowid_column_index]; + UnifiedVectorFormat rowid_data; + rowid_vec.ToUnifiedFormat(input.size(), rowid_data); + auto rowids = UnifiedVectorFormat::GetData(rowid_data); + + unordered_set observed_vectors; + observed_vectors.reserve(input.size()); + for (idx_t input_idx = 0; input_idx < input.size(); ++input_idx) { + auto rowid_idx = rowid_data.sel->get_index(input_idx); + if (!rowid_data.validity.RowIsValid(rowid_idx)) { + continue; + } + idx_t key; + if (TryEncodeVectorKey(rowids[rowid_idx], key)) { + observed_vectors.insert(key); + } + } + + if (observed_vectors.empty()) { + return OperatorResultType::NEED_MORE_INPUT; + } + + SelectionVector sel(input.size()); + idx_t match_count = local_state.expr_executor.SelectExpression(input, sel); + unordered_set matching_vectors; + matching_vectors.reserve(match_count); + for (idx_t match_idx = 0; match_idx < match_count; ++match_idx) { + auto input_idx = sel.get_index(match_idx); + auto rowid_idx = rowid_data.sel->get_index(input_idx); + if (!rowid_data.validity.RowIsValid(rowid_idx)) { + continue; + } + idx_t key; + if (TryEncodeVectorKey(rowids[rowid_idx], key)) { + matching_vectors.insert(key); + } + } + + for (const auto &key : observed_vectors) { + auto rg_idx = key / VECTORS_PER_ROW_GROUP; + auto vec_idx = key % VECTORS_PER_ROW_GROUP; + RecordChunkObservation(*local_state.local_entry, rg_idx, vec_idx, matching_vectors.count(key) > 0); + } + + return OperatorResultType::NEED_MORE_INPUT; +} + +void PhysicalCacheRecorder::RecordChunkObservation(ConditionCacheEntry &local_entry, idx_t rg_idx, idx_t vec_idx, + bool has_qualifying) { + local_entry.EnsureRowGroup(rg_idx); + local_entry.SetObservedVector(rg_idx, vec_idx); + if (has_qualifying) { + local_entry.SetQualifyingVector(rg_idx, vec_idx); + } +} + +namespace { + +void BackfillMissingObservations(ClientContext &context, ConditionCacheEntry &destination, const string &table_catalog, + const string &table_schema, const string &table_name, Expression &predicate) { + if (table_catalog.empty() || table_schema.empty() || table_name.empty()) { + return; + } + auto &table_entry = Catalog::GetEntry(context, table_catalog, table_schema, table_name); + auto ranges = destination.GetUnobservedVectorRanges(table_entry.GetStorage().GetTotalRows()); + if (ranges.empty()) { + return; + } + auto backfill_entry = BuildCacheEntryForRanges(context, table_entry, predicate, ranges); + destination.MergeFrom(*backfill_entry); +} + +} // namespace + +OperatorFinalResultType PhysicalCacheRecorder::OperatorFinalize(Pipeline &pipeline, Event &event, + ClientContext &context, + OperatorFinalizeInput &input) const { + auto &global_state = input.global_state.Cast(); + auto store = ConditionCacheStore::GetOrCreate(context); + CacheKey key {table_oid, canonical_key}; + + auto destination = cache_entry ? cache_entry : make_shared_ptr(); + if (metadata_entry) { + destination->MergeFrom(*metadata_entry); + } + { + concurrency::lock_guard guard(global_state.lock); + for (const auto &task_entry : global_state.task_local_entries) { + destination->MergeFrom(*task_entry); + } + } + + auto existing = store->Lookup(context, key); + if (existing && existing.get() != destination.get()) { + destination->MergeFrom(*existing); + } + + if (backfill_predicate) { + try { + BackfillMissingObservations(context, *destination, table_catalog, table_schema, table_name, + *backfill_predicate); + } catch (...) { + // Backfill is an optimization side effect. Preserve query correctness if it cannot run. + } + } + + store->Upsert(context, key, destination); + return OperatorFinalResultType::FINISHED; +} + +bool PhysicalCacheRecorder::RequiresOperatorFinalize() const { + return true; +} + +bool PhysicalCacheRecorder::ParallelOperator() const { + return true; +} + +string PhysicalCacheRecorder::GetName() const { + return "CACHE_RECORDER"; +} + +InsertionOrderPreservingMap PhysicalCacheRecorder::ParamsToString() const { + InsertionOrderPreservingMap result; + result["Table OID"] = to_string(table_oid); + result["Filter Key"] = canonical_key; + result["Row ID Column"] = to_string(rowid_column_index); + return result; +} + +} // namespace duckdb diff --git a/src/query_condition_cache_extension.cpp b/src/query_condition_cache_extension.cpp index a79d832..a8a8451 100644 --- a/src/query_condition_cache_extension.cpp +++ b/src/query_condition_cache_extension.cpp @@ -6,7 +6,9 @@ #include "duckdb/main/config.hpp" #include "duckdb/main/extension/extension_loader.hpp" #include "duckdb/optimizer/optimizer_extension.hpp" +#include "duckdb/planner/extension_callback.hpp" #include "cache_invalidation_optimizer.hpp" +#include "logical_cache_recorder.hpp" #include "logical_cache_invalidator.hpp" #include "query_condition_cache_filter.hpp" #include "query_condition_cache_functions.hpp" @@ -42,6 +44,7 @@ void LoadInternal(ExtensionLoader &loader) { // Register optimizer extension OptimizerExtension::Register(config, QueryConditionCacheOptimizer()); OptimizerExtension::Register(config, CacheInvalidationOptimizer()); + OperatorExtension::Register(config, make_shared_ptr()); OperatorExtension::Register(config, make_shared_ptr()); } } // namespace diff --git a/src/query_condition_cache_functions.cpp b/src/query_condition_cache_functions.cpp index 1792300..5692e7a 100644 --- a/src/query_condition_cache_functions.cpp +++ b/src/query_condition_cache_functions.cpp @@ -97,6 +97,18 @@ struct ScanColumn { LogicalType type; }; +void MarkObservedRows(ConditionCacheEntry &entry, idx_t start_row, idx_t count) { + idx_t row = start_row; + idx_t end = start_row + count; + while (row < end) { + idx_t rg_idx = row / DEFAULT_ROW_GROUP_SIZE; + idx_t vec_idx = (row % DEFAULT_ROW_GROUP_SIZE) / STANDARD_VECTOR_SIZE; + entry.SetObservedVector(rg_idx, vec_idx); + idx_t next_vector = (rg_idx * DEFAULT_ROW_GROUP_SIZE) + ((vec_idx + 1) * STANDARD_VECTOR_SIZE); + row = MinValue(next_vector, end); + } +} + // Task that scans a subset of row groups in parallel and builds a local ConditionCacheEntry. class CacheBuildTask : public BaseExecutorTask { public: @@ -254,6 +266,49 @@ shared_ptr BuildCacheEntry(ClientContext &context, DuckTabl auto entry = make_shared_ptr(); MergeLocalCacheEntries(local_entries, entry); + entry->MarkAllRowGroupsFullyObserved(); + + return entry; +} + +shared_ptr BuildCacheEntryForRanges(ClientContext &context, DuckTableEntry &table_entry, + Expression &bound_expr, + const vector &ranges) { + auto entry = make_shared_ptr(); + if (ranges.empty()) { + return entry; + } + + auto &storage = table_entry.GetStorage(); + auto &columns = table_entry.GetColumns(); + + unordered_map storage_to_scan_idx; + idx_t scan_pos = 0; + for (const auto &col : columns.Physical()) { + storage_to_scan_idx[col.Oid()] = scan_pos++; + } + + RemapColumnIndices(bound_expr, storage_to_scan_idx); + + auto &transaction = DuckTransaction::Get(context, table_entry.ParentCatalog().GetAttached()); + ExpressionExecutor expr_executor(context, bound_expr); + + for (const auto &range : ranges) { + idx_t current_row = range.start_row; + storage.ScanTableSegment(transaction, range.start_row, range.count, [&](DataChunk &chunk) { + MarkObservedRows(*entry, current_row, chunk.size()); + + SelectionVector sel(chunk.size()); + idx_t match_count = expr_executor.SelectExpression(chunk, sel); + for (idx_t idx = 0; idx < match_count; ++idx) { + idx_t row_id = current_row + sel.get_index(idx); + idx_t rg_idx = row_id / DEFAULT_ROW_GROUP_SIZE; + idx_t vector_idx = (row_id % DEFAULT_ROW_GROUP_SIZE) / STANDARD_VECTOR_SIZE; + entry->SetQualifyingVector(/*rg_idx=*/rg_idx, /*vec_idx=*/vector_idx); + } + current_row += chunk.size(); + }); + } return entry; } @@ -290,7 +345,7 @@ void ConditionCacheBuildExecute(ClientContext &context, TableFunctionInput &data output.SetCardinality(1); output.data[0].SetValue(0, StringUtil::Format("Cache Built: %llu/%llu vectors, %llu/%llu row groups", stats.qualifying_vectors, stats.total_vectors, - stats.qualifying_row_groups, stats.total_row_groups)); + stats.cached_row_groups, stats.total_row_groups)); } TableFunction ConditionCacheBuildFunction() { @@ -361,7 +416,7 @@ void ConditionCacheInfoExecute(ClientContext &context, TableFunctionInput &data_ output.SetCardinality(1); if (entry) { auto stats = entry->ComputeStats(bind_data.total_rows); - output.data[0].SetValue(0, Value::INTEGER(static_cast(stats.qualifying_row_groups))); + output.data[0].SetValue(0, Value::INTEGER(static_cast(stats.cached_row_groups))); output.data[1].SetValue(0, Value::INTEGER(static_cast(stats.total_row_groups))); output.data[2].SetValue(0, Value::INTEGER(static_cast(stats.qualifying_vectors))); output.data[3].SetValue(0, Value::INTEGER(static_cast(stats.total_vectors))); diff --git a/src/query_condition_cache_optimizer.cpp b/src/query_condition_cache_optimizer.cpp index d531da1..1b35067 100644 --- a/src/query_condition_cache_optimizer.cpp +++ b/src/query_condition_cache_optimizer.cpp @@ -1,12 +1,15 @@ #include "query_condition_cache_optimizer.hpp" -#include "query_condition_cache_filter.hpp" +#include "logical_cache_recorder.hpp" #include "predicate_key_utils.hpp" -#include "query_condition_cache_functions.hpp" +#include "query_condition_cache_filter.hpp" #include "query_condition_cache_state.hpp" #include "duckdb/catalog/catalog_entry/duck_table_entry.hpp" +#include "duckdb/common/assert.hpp" +#include "duckdb/common/numeric_utils.hpp" #include "duckdb/common/vector.hpp" +#include "duckdb/function/partition_stats.hpp" #include "duckdb/planner/expression/bound_cast_expression.hpp" #include "duckdb/planner/expression/bound_columnref_expression.hpp" #include "duckdb/planner/expression/bound_function_expression.hpp" @@ -17,6 +20,31 @@ namespace duckdb { +namespace { + +bool ConvertColumnRefsToStorageRefs(unique_ptr &expr, const LogicalGet &get) { + if (expr->GetExpressionClass() == ExpressionClass::BOUND_COLUMN_REF) { + auto &colref = expr->Cast(); + auto &column_ids = get.GetColumnIds(); + if (colref.binding.column_index >= column_ids.size()) { + return false; + } + StorageIndex storage_index; + if (!get.TryGetStorageIndex(column_ids[colref.binding.column_index], storage_index)) { + return false; + } + expr = make_uniq(colref.alias, colref.return_type, storage_index.GetPrimaryIndex()); + return true; + } + + bool success = true; + ExpressionIterator::EnumerateChildren( + *expr, [&](unique_ptr &child) { success = ConvertColumnRefsToStorageRefs(child, get) && success; }); + return success; +} + +} // namespace + QueryConditionCacheOptimizer::QueryConditionCacheOptimizer() { pre_optimize_function = PreOptimizeFunction; optimize_function = OptimizeFunction; @@ -31,9 +59,6 @@ bool QueryConditionCacheOptimizer::IsSettingEnabled(ClientContext &context) { return val.GetValue(); } -// Pre-optimize runs BEFORE built-in passes (including FilterPushdown). -// The full WHERE clause is still in LogicalFilter.expressions, so we can -// compute a canonical key that covers the entire predicate. void QueryConditionCacheOptimizer::PreOptimizeFunction(OptimizerExtensionInput &input, unique_ptr &plan) { if (!IsSettingEnabled(input.context)) { @@ -42,32 +67,35 @@ void QueryConditionCacheOptimizer::PreOptimizeFunction(OptimizerExtensionInput & auto query_state = input.context.registered_state->GetOrCreate(CacheOptimizerQueryState::NAME); query_state->cache_apply_pending.clear(); + query_state->cache_recorder_pending.clear(); try { - PreOptimizeWalk(input.context, plan, /*inside_dml=*/false, *query_state); + PreOptimizeWalk(input.context, plan, /*inside_dml=*/false, /*inside_truncating=*/false, *query_state); } catch (...) { - // Defense in depth: skip cache optimization rather than failing the query. query_state->cache_apply_pending.clear(); + query_state->cache_recorder_pending.clear(); } } void QueryConditionCacheOptimizer::PreOptimizeWalk(ClientContext &context, unique_ptr &plan, - bool inside_dml, CacheOptimizerQueryState &state) { - // Skip cache building inside DML subplans + bool inside_dml, bool inside_truncating, + CacheOptimizerQueryState &state) { bool is_dml = plan->type == LogicalOperatorType::LOGICAL_DELETE || plan->type == LogicalOperatorType::LOGICAL_UPDATE || plan->type == LogicalOperatorType::LOGICAL_INSERT || plan->type == LogicalOperatorType::LOGICAL_MERGE_INTO; + bool is_truncating = plan->type == LogicalOperatorType::LOGICAL_LIMIT || + plan->type == LogicalOperatorType::LOGICAL_TOP_N || + plan->type == LogicalOperatorType::LOGICAL_SAMPLE; bool child_inside_dml = inside_dml || is_dml; + bool child_inside_truncating = inside_truncating || is_truncating; for (auto &child : plan->children) { - PreOptimizeWalk(context, child, child_inside_dml, state); + PreOptimizeWalk(context, child, child_inside_dml, child_inside_truncating, state); } - if (inside_dml) { + if (inside_dml || inside_truncating) { return; } - // Only handle direct table scans with a filter (LogicalFilter -> LogicalGet). - // Joins, subqueries, and other patterns are not supported currently. if (plan->type != LogicalOperatorType::LOGICAL_FILTER || plan->children.empty()) { return; } @@ -78,21 +106,17 @@ void QueryConditionCacheOptimizer::PreOptimizeWalk(ClientContext &context, uniqu auto &filter = plan->Cast(); auto &get = plan->children[0]->Cast(); auto table = get.GetTable(); - if (!table) { - return; // not a DuckDB table (e.g. system table, external) - } - if (filter.expressions.empty()) { + if (!table || filter.expressions.empty()) { return; } auto &duck_table = table->Cast(); auto &storage = duck_table.GetStorage(); - - // Skip caching on indexed tables: our extra ROW_ID filter would force - // filter_set.filters.size() > 1, disabling the ART index scan path. if (storage.HasIndexes()) { return; } + idx_t total_rows = storage.GetTotalRows(); + idx_t table_index = get.table_index; CacheKey key {table->oid, ComputeCanonicalPredicateKey(filter.expressions)}; if (key.filter_key.empty()) { @@ -102,63 +126,49 @@ void QueryConditionCacheOptimizer::PreOptimizeWalk(ClientContext &context, uniqu auto store = ConditionCacheStore::GetOrCreate(context); auto entry = store->Lookup(context, key); + bool should_inject_recorder = false; if (!entry) { - // TODO: Consider building cache in the background and syncing later - // to avoid blocking the first query. - entry = BuildCacheForPredicate(context, filter.expressions, get); - if (entry) { - store->Upsert(context, key, entry); - } - } - - if (entry) { - state.cache_apply_pending[get.table_index] = std::move(entry); - } -} - -namespace { - -// Rewrite BoundColumnRefExpression -> BoundReferenceExpression(storage OID), -// the shape BuildCacheEntry expects. The source column_index is a position -// into LogicalGet::column_ids, which maps to the storage OID. -void ConvertColumnRefsToScanRefs(unique_ptr &expr, const LogicalGet &get) { - if (expr->GetExpressionClass() == ExpressionClass::BOUND_COLUMN_REF) { - auto &colref = expr->Cast(); - auto &column_ids = get.GetColumnIds(); - auto col_idx = colref.binding.column_index; - ALWAYS_ASSERT(col_idx < column_ids.size()); - auto storage_oid = column_ids[col_idx].GetPrimaryIndex(); - expr = make_uniq(colref.alias, colref.return_type, storage_oid); - return; - } - ExpressionIterator::EnumerateChildren( - *expr, [&](unique_ptr &child) { ConvertColumnRefsToScanRefs(child, get); }); -} - -} // namespace - -shared_ptr QueryConditionCacheOptimizer::BuildCacheForPredicate( - ClientContext &context, const vector> &expressions, LogicalGet &get) { - auto table_ptr = get.GetTable(); - if (!table_ptr) { - return nullptr; + entry = make_shared_ptr(); + should_inject_recorder = true; + } else if (entry->NeedsObservation(total_rows)) { + should_inject_recorder = true; } - auto &table_entry = table_ptr->Cast(); - // Clone the plan's already-bound filter expressions and remap column refs - // to storage OIDs. - vector> cloned; - cloned.reserve(expressions.size()); - for (const auto &expr : expressions) { - auto copy = expr->Copy(); - ConvertColumnRefsToScanRefs(copy, get); - cloned.push_back(std::move(copy)); + if (should_inject_recorder) { + vector> cloned; + cloned.reserve(filter.expressions.size()); + vector> backfill_cloned; + backfill_cloned.reserve(filter.expressions.size()); + bool can_backfill = true; + for (const auto &expr : filter.expressions) { + cloned.push_back(expr->Copy()); + auto backfill_copy = expr->Copy(); + can_backfill = ConvertColumnRefsToStorageRefs(backfill_copy, get) && can_backfill; + backfill_cloned.push_back(std::move(backfill_copy)); + } + auto predicate = CombineWithAnd(std::move(cloned)); + predicate = + BoundCastExpression::AddCastToType(context, std::move(predicate), LogicalType {LogicalTypeId::BOOLEAN}); + unique_ptr backfill_predicate; + if (can_backfill) { + backfill_predicate = CombineWithAnd(std::move(backfill_cloned)); + backfill_predicate = BoundCastExpression::AddCastToType(context, std::move(backfill_predicate), + LogicalType {LogicalTypeId::BOOLEAN}); + } + state.cache_recorder_pending[table_index] = RecorderInjectionInfo { + table->oid, + table->ParentCatalog().GetName(), + table->ParentSchema().name, + table->name, + key.filter_key, + std::move(predicate), + std::move(backfill_predicate), + entry, + nullptr, + }; } - auto predicate = CombineWithAnd(std::move(cloned)); - predicate = BoundCastExpression::AddCastToType(context, std::move(predicate), LogicalType {LogicalTypeId::BOOLEAN}); - - return BuildCacheEntry(context, table_entry, *predicate); + state.cache_apply_pending[table_index] = entry; } void QueryConditionCacheOptimizer::PostOptimizeWalk(ClientContext &context, unique_ptr &plan, @@ -167,18 +177,27 @@ void QueryConditionCacheOptimizer::PostOptimizeWalk(ClientContext &context, uniq PostOptimizeWalk(context, child, state); } - if (plan->type != LogicalOperatorType::LOGICAL_GET) { - return; - } + if (plan->type == LogicalOperatorType::LOGICAL_GET) { + auto &get = plan->Cast(); + auto entry = state.cache_apply_pending.find(get.table_index); + if (entry == state.cache_apply_pending.end()) { + return; + } - auto &get = plan->Cast(); - auto entry = state.cache_apply_pending.find(get.table_index); - if (entry == state.cache_apply_pending.end()) { - return; - } + InjectCacheFilter(context, get, entry->second); + state.cache_apply_pending.erase(entry); + auto recorder_it = state.cache_recorder_pending.find(get.table_index); + if (recorder_it == state.cache_recorder_pending.end()) { + return; + } - InjectCacheFilter(context, get, entry->second); - state.cache_apply_pending.erase(entry); + auto info = std::move(recorder_it->second); + state.cache_recorder_pending.erase(recorder_it); + info.metadata_entry = GetPrunedRowGroupsFromTableFilters(context, get); + InjectCacheRecorder(plan, info.table_oid, std::move(info.canonical_key), std::move(info.table_catalog), + std::move(info.table_schema), std::move(info.table_name), std::move(info.predicate), + std::move(info.backfill_predicate), info.entry, info.metadata_entry); + } } void QueryConditionCacheOptimizer::InjectCacheFilter(ClientContext &context, LogicalGet &get, @@ -212,6 +231,116 @@ void QueryConditionCacheOptimizer::InjectCacheFilter(ClientContext &context, Log make_uniq(std::move(filter_expr), entry)); } +void QueryConditionCacheOptimizer::InjectCacheRecorder(unique_ptr &plan, idx_t table_oid, + string canonical_key, string table_catalog, string table_schema, + string table_name, unique_ptr predicate, + unique_ptr backfill_predicate, + const shared_ptr &entry, + const shared_ptr &metadata_entry) { + D_ASSERT(plan->type == LogicalOperatorType::LOGICAL_GET); + auto &get = plan->Cast(); + idx_t rowid_chunk_idx = EnsureRowIdChunkIndex(get); + auto recorder = make_uniq( + table_oid, std::move(canonical_key), std::move(predicate), rowid_chunk_idx, std::move(table_catalog), + std::move(table_schema), std::move(table_name), std::move(backfill_predicate), entry, metadata_entry); + recorder->children.push_back(std::move(plan)); + plan = std::move(recorder); +} + +shared_ptr +QueryConditionCacheOptimizer::GetPrunedRowGroupsFromTableFilters(ClientContext &context, const LogicalGet &get) { + if (get.table_filters.filters.empty()) { + return nullptr; + } + + auto table = get.GetTable(); + if (!table) { + return nullptr; + } + + vector> pushed_filters; + pushed_filters.reserve(get.table_filters.filters.size()); + for (const auto &filter_entry : get.table_filters.filters) { + StorageIndex storage_index; + if (!get.TryGetStorageIndex(ColumnIndex(filter_entry.first), storage_index)) { + continue; + } + pushed_filters.emplace_back(storage_index, filter_entry.second.get()); + } + if (pushed_filters.empty()) { + return nullptr; + } + + auto &duck_table = table->Cast(); + auto &storage = duck_table.GetStorage(); + auto metadata_entry = make_shared_ptr(); + auto partition_stats = storage.GetPartitionStats(context); + for (auto &partition : partition_stats) { + if (!partition.row_start.IsValid() || !partition.partition_row_group) { + continue; + } + auto row_start = partition.row_start.GetIndex(); + if (row_start >= NumericCast(MAX_ROW_ID)) { + continue; + } + + bool row_group_pruned = false; + for (const auto &filter : pushed_filters) { + auto column_stats = partition.partition_row_group->GetColumnStatistics(filter.first); + if (!column_stats) { + continue; + } + if (filter.second->CheckStatistics(*column_stats) == FilterPropagateResult::FILTER_ALWAYS_FALSE) { + row_group_pruned = true; + break; + } + } + if (row_group_pruned) { + metadata_entry->MarkRowGroupFullyObserved(row_start / DEFAULT_ROW_GROUP_SIZE); + } + } + + if (metadata_entry->RowGroupCount() == 0) { + return nullptr; + } + return metadata_entry; +} + +idx_t QueryConditionCacheOptimizer::EnsureRowIdChunkIndex(LogicalGet &get) { + auto &column_ids = get.GetMutableColumnIds(); + idx_t rowid_column_ids_pos = column_ids.size(); + for (idx_t i = 0; i < column_ids.size(); ++i) { + if (column_ids[i].IsRowIdColumn()) { + rowid_column_ids_pos = i; + break; + } + } + if (rowid_column_ids_pos == column_ids.size()) { + if (get.projection_ids.empty() && !column_ids.empty()) { + get.projection_ids.reserve(column_ids.size() + 1); + for (idx_t i = 0; i < column_ids.size(); i++) { + get.projection_ids.push_back(i); + } + } + column_ids.emplace_back(COLUMN_IDENTIFIER_ROW_ID); + rowid_column_ids_pos = column_ids.size() - 1; + } + + if (get.projection_ids.empty()) { + get.projection_ids.reserve(column_ids.size()); + for (idx_t i = 0; i < column_ids.size(); ++i) { + get.projection_ids.push_back(i); + } + } + for (idx_t i = 0; i < get.projection_ids.size(); ++i) { + if (get.projection_ids[i] == rowid_column_ids_pos) { + return i; + } + } + get.projection_ids.push_back(rowid_column_ids_pos); + return get.projection_ids.size() - 1; +} + void QueryConditionCacheOptimizer::OptimizeFunction(OptimizerExtensionInput &input, unique_ptr &plan) { if (!IsSettingEnabled(input.context)) { return; @@ -223,6 +352,8 @@ void QueryConditionCacheOptimizer::OptimizeFunction(OptimizerExtensionInput &inp } PostOptimizeWalk(input.context, plan, *query_state); + query_state->cache_apply_pending.clear(); + query_state->cache_recorder_pending.clear(); } } // namespace duckdb diff --git a/src/query_condition_cache_state.cpp b/src/query_condition_cache_state.cpp index 1844ba5..5d07273 100644 --- a/src/query_condition_cache_state.cpp +++ b/src/query_condition_cache_state.cpp @@ -10,31 +10,41 @@ namespace duckdb { RowGroupFilter::RowGroupFilter(const vector &qualifying_vectors) { for (const auto &vec_idx : qualifying_vectors) { - matching_vectors.at(vec_idx / 64) |= (1ULL << (vec_idx % 64)); + matching_vectors.set(vec_idx); } } void RowGroupFilter::SetVector(idx_t vector_index) { - matching_vectors.at(vector_index / 64) |= (1ULL << (vector_index % 64)); + matching_vectors.set(vector_index); } bool RowGroupFilter::VectorHasRows(idx_t vector_index) const { - return (matching_vectors.at(vector_index / 64) >> (vector_index % 64)) & 1ULL; + return matching_vectors.test(vector_index); } bool RowGroupFilter::IsEmpty() const { - for (const auto &w : matching_vectors) { - if (w != 0) { - return false; - } - } - return true; + return matching_vectors.none(); +} + +void RowGroupFilter::SetObserved(idx_t vector_index) { + observed.set(vector_index); +} + +void RowGroupFilter::MarkFullyObserved() { + observed.set(); +} + +bool RowGroupFilter::IsObserved(idx_t vector_index) const { + return observed.test(vector_index); +} + +bool RowGroupFilter::IsFullyObserved() const { + return observed.all(); } void RowGroupFilter::MergeFrom(const RowGroupFilter &other) { - for (idx_t i = 0; i < BITVECTOR_ARRAY_SIZE; ++i) { - matching_vectors[i] |= other.matching_vectors[i]; - } + matching_vectors |= other.matching_vectors; + observed |= other.observed; } // ------- CONDITION_CACHE_ENTRY ------- @@ -51,17 +61,10 @@ CacheEntryStats ConditionCacheEntry::ComputeStats(idx_t total_rows) const { constexpr idx_t vectors_per_row_group = DEFAULT_ROW_GROUP_SIZE / STANDARD_VECTOR_SIZE; idx_t qualifying_vectors = 0; - idx_t qualifying_row_groups = 0; for (const auto &[rg_idx, filter] : bitvectors) { - if (!filter.IsEmpty()) { - ++qualifying_row_groups; - } - for (idx_t v = 0; v < vectors_per_row_group; ++v) { - if (filter.VectorHasRows(v)) { - ++qualifying_vectors; - } - } + qualifying_vectors += filter.matching_vectors.count(); } + idx_t cached_row_groups = bitvectors.size(); idx_t full_row_groups = total_rows / DEFAULT_ROW_GROUP_SIZE; idx_t remaining_rows = total_rows % DEFAULT_ROW_GROUP_SIZE; @@ -74,7 +77,7 @@ CacheEntryStats ConditionCacheEntry::ComputeStats(idx_t total_rows) const { return CacheEntryStats { .qualifying_vectors = qualifying_vectors, .total_vectors = total_vectors, - .qualifying_row_groups = qualifying_row_groups, + .cached_row_groups = cached_row_groups, .total_row_groups = total_row_groups, }; } @@ -107,12 +110,32 @@ void ConditionCacheEntry::MergeFrom(const ConditionCacheEntry &other) { } } +void ConditionCacheEntry::MarkAllRowGroupsFullyObserved() { + concurrency::lock_guard guard(lock); + for (auto &[rg_idx, filter] : bitvectors) { + filter.MarkFullyObserved(); + } +} + +void ConditionCacheEntry::MarkRowGroupFullyObserved(idx_t rg_idx) { + concurrency::lock_guard guard(lock); + bitvectors[rg_idx].MarkFullyObserved(); +} + +void ConditionCacheEntry::SetObservedVector(idx_t rg_idx, idx_t vec_idx) { + concurrency::lock_guard guard(lock); + bitvectors[rg_idx].SetObserved(vec_idx); +} + bool ConditionCacheEntry::VectorPassesFilter(idx_t rg_idx, idx_t vec_idx) const { concurrency::lock_guard guard(lock); auto it = bitvectors.find(rg_idx); if (it == bitvectors.end()) { return true; } + if (!it->second.IsObserved(vec_idx)) { + return true; + } return it->second.VectorHasRows(vec_idx); } @@ -123,10 +146,74 @@ bool ConditionCacheEntry::StatisticsRangeIsAllEmptyCached(idx_t min_rg, idx_t ma if (it == bitvectors.end() || !it->second.IsEmpty()) { return false; } + if (!it->second.IsFullyObserved()) { + return false; + } } return true; } +bool ConditionCacheEntry::NeedsObservation(idx_t total_rows) const { + concurrency::lock_guard guard(lock); + idx_t total_row_groups = (total_rows + DEFAULT_ROW_GROUP_SIZE - 1) / DEFAULT_ROW_GROUP_SIZE; + for (idx_t rg = 0; rg < total_row_groups; ++rg) { + auto it = bitvectors.find(rg); + if (it == bitvectors.end()) { + return true; + } + idx_t rows_in_rg = DEFAULT_ROW_GROUP_SIZE; + if (rg + 1 == total_row_groups) { + rows_in_rg = total_rows - rg * DEFAULT_ROW_GROUP_SIZE; + if (rows_in_rg == 0) { + rows_in_rg = DEFAULT_ROW_GROUP_SIZE; + } + } + idx_t vectors_in_rg = (rows_in_rg + STANDARD_VECTOR_SIZE - 1) / STANDARD_VECTOR_SIZE; + for (idx_t vec_idx = 0; vec_idx < vectors_in_rg; ++vec_idx) { + if (!it->second.IsObserved(vec_idx)) { + return true; + } + } + } + return false; +} + +vector ConditionCacheEntry::GetUnobservedVectorRanges(idx_t total_rows) const { + concurrency::lock_guard guard(lock); + vector ranges; + idx_t total_row_groups = (total_rows + DEFAULT_ROW_GROUP_SIZE - 1) / DEFAULT_ROW_GROUP_SIZE; + + auto append_range = [&](idx_t start_row, idx_t count) { + if (count == 0) { + return; + } + if (!ranges.empty()) { + auto &last = ranges.back(); + if (last.start_row + last.count == start_row) { + last.count += count; + return; + } + } + ranges.push_back(CacheObservationRange {.start_row = start_row, .count = count}); + }; + + for (idx_t rg = 0; rg < total_row_groups; ++rg) { + idx_t row_group_start = rg * DEFAULT_ROW_GROUP_SIZE; + idx_t rows_in_rg = MinValue(DEFAULT_ROW_GROUP_SIZE, total_rows - row_group_start); + idx_t vectors_in_rg = (rows_in_rg + STANDARD_VECTOR_SIZE - 1) / STANDARD_VECTOR_SIZE; + auto it = bitvectors.find(rg); + for (idx_t vec_idx = 0; vec_idx < vectors_in_rg; ++vec_idx) { + if (it != bitvectors.end() && it->second.IsObserved(vec_idx)) { + continue; + } + idx_t vector_start = row_group_start + vec_idx * STANDARD_VECTOR_SIZE; + idx_t vector_count = MinValue(STANDARD_VECTOR_SIZE, total_rows - vector_start); + append_range(vector_start, vector_count); + } + } + return ranges; +} + idx_t ConditionCacheEntry::RowGroupCount() const { concurrency::lock_guard guard(lock); return bitvectors.size(); @@ -137,6 +224,15 @@ bool ConditionCacheEntry::HasRowGroup(idx_t rg_idx) const { return bitvectors.find(rg_idx) != bitvectors.end(); } +idx_t ConditionCacheEntry::GetObservedVectorCount(idx_t rg_idx) const { + concurrency::lock_guard guard(lock); + auto it = bitvectors.find(rg_idx); + if (it == bitvectors.end()) { + return 0; + } + return static_cast(it->second.observed.count()); +} + bool ConditionCacheEntry::RowGroupVectorHasQualifyingRows(idx_t rg_idx, idx_t vec_idx) const { concurrency::lock_guard guard(lock); auto it = bitvectors.find(rg_idx); @@ -164,6 +260,20 @@ pair ConditionCacheEntry::EraseRowGroups(const unordered_set return {removed, bitvectors.empty()}; } +pair ConditionCacheEntry::EraseRowGroupsStartingAt(idx_t first_row_group) { + concurrency::lock_guard guard(lock); + idx_t removed = 0; + for (auto it = bitvectors.begin(); it != bitvectors.end();) { + if (it->first >= first_row_group) { + it = bitvectors.erase(it); + removed++; + } else { + ++it; + } + } + return {removed, bitvectors.empty()}; +} + // ------- TABLE_FILTER_KEY_INDEX ------- void TableFilterKeyIndex::Add(const string &filter_key) { @@ -253,6 +363,80 @@ idx_t ConditionCacheStore::RemoveRowGroupsForTable(ClientContext &context, idx_t return removed_count; } +idx_t ConditionCacheStore::RemoveRowGroupsStartingAtForTable(ClientContext &context, idx_t table_oid, + idx_t first_row_group) { + auto &cache = ObjectCache::GetObjectCache(context); + + auto index = cache.Get(MakeFilterKeyIndexKey(table_oid)); + if (!index) { + return 0; + } + + auto filter_keys = index->Take(); + idx_t removed_count = 0; + + for (const auto &filter_key : filter_keys) { + CacheKey key {table_oid, filter_key}; + string cache_key = MakeCacheKeyString(key); + auto entry = cache.Get(cache_key); + if (!entry) { + continue; + } + auto erased = entry->EraseRowGroupsStartingAt(first_row_group); + removed_count += erased.first; + if (erased.second) { + cache.Delete(cache_key); + } else { + index->Add(filter_key); + } + } + + if (index->IsEmpty()) { + cache.Delete(MakeFilterKeyIndexKey(table_oid)); + concurrency::lock_guard guard(lock); + cached_table_oids.erase(table_oid); + } + + return removed_count; +} + +idx_t ConditionCacheStore::RemoveRowGroupsStartingAtForTable(DatabaseInstance &db, idx_t table_oid, + idx_t first_row_group) { + auto &cache = db.GetObjectCache(); + + auto index = cache.Get(MakeFilterKeyIndexKey(table_oid)); + if (!index) { + return 0; + } + + auto filter_keys = index->Take(); + idx_t removed_count = 0; + + for (const auto &filter_key : filter_keys) { + CacheKey key {table_oid, filter_key}; + string cache_key = MakeCacheKeyString(key); + auto entry = cache.Get(cache_key); + if (!entry) { + continue; + } + auto erased = entry->EraseRowGroupsStartingAt(first_row_group); + removed_count += erased.first; + if (erased.second) { + cache.Delete(cache_key); + } else { + index->Add(filter_key); + } + } + + if (index->IsEmpty()) { + cache.Delete(MakeFilterKeyIndexKey(table_oid)); + concurrency::lock_guard guard(lock); + cached_table_oids.erase(table_oid); + } + + return removed_count; +} + bool ConditionCacheStore::HasEntriesForTable(ClientContext &context, idx_t table_oid) { auto &cache = ObjectCache::GetObjectCache(context); auto index = cache.Get(MakeFilterKeyIndexKey(table_oid)); @@ -282,4 +466,8 @@ shared_ptr ConditionCacheStore::GetOrCreate(ClientContext & return cache.GetOrCreate(CACHE_KEY); } +shared_ptr ConditionCacheStore::GetOrCreate(DatabaseInstance &db) { + return db.GetObjectCache().GetOrCreate(CACHE_KEY); +} + } // namespace duckdb diff --git a/test/sql/condition_cache_auto.test b/test/sql/condition_cache_auto.test index aac71d2..e230eaa 100644 --- a/test/sql/condition_cache_auto.test +++ b/test/sql/condition_cache_auto.test @@ -24,7 +24,7 @@ SELECT count(*) FROM t WHERE id < 3000; query IIII SELECT * FROM condition_cache_info('t', 'id < 3000'); ---- -1 5 2 245 +5 5 2 245 # Disable setting, new predicates should not be cached statement ok diff --git a/test/sql/condition_cache_build.test b/test/sql/condition_cache_build.test index c51ba9c..ff1ade7 100644 --- a/test/sql/condition_cache_build.test +++ b/test/sql/condition_cache_build.test @@ -18,7 +18,7 @@ Cache Built: 245/245 vectors, 5/5 row groups query I SELECT status FROM condition_cache_build('t', 'id < 3000'); ---- -Cache Built: 2/245 vectors, 1/5 row groups +Cache Built: 2/245 vectors, 5/5 row groups # Build same predicate again (upsert, no error) query I diff --git a/test/sql/condition_cache_explain.test b/test/sql/condition_cache_explain.test new file mode 100644 index 0000000..9347f46 --- /dev/null +++ b/test/sql/condition_cache_explain.test @@ -0,0 +1,51 @@ +# name: test/sql/condition_cache_explain.test +# description: Test condition cache plan injection with explain +# group: [sql] + +require query_condition_cache + +statement ok +PRAGMA explain_output = PHYSICAL_ONLY; + +statement ok +CREATE TABLE t AS SELECT i AS id, i % 100 AS val FROM range(1000) t(i); + +# Cache miss: inject both the recorder and the cache filter. +query TT +EXPLAIN SELECT count(*) FROM t WHERE val = 42; +---- +physical_plan :.*UNGROUPED_AGGREGATE.*Aggregates:.*count_star\(\).*CACHE_RECORDER.*Table OID: [0-9]+.*Filter Key:.*\(val = CAST\(42 AS BIGINT\)\).*Row ID Column: 1.*SEQ_SCAN.*Table: memory\.main\.t.*Type: Sequential Scan.*Projections: val.*Filters:.*val=42.*__condition_cache_filter.*\(rowid\).*~11 rows.* + +statement ok +SELECT count(*) FROM t WHERE val = 42; + +# Cache hit: keep the cache filter, but do not re-inject the recorder. +query TT +EXPLAIN SELECT count(*) FROM t WHERE val = 42; +---- +physical_plan :.*UNGROUPED_AGGREGATE.*Aggregates:.*count_star\(\).*SEQ_SCAN.*Table: memory\.main\.t.*Type: Sequential Scan.*Projections: val.*Filters:.*val=42.*__condition_cache_filter.*\(rowid\).*~11 rows.* + +# Cache hit with new row groups: re-inject the recorder so missing coverage can be extended. +statement ok +CREATE TABLE t_full AS SELECT i AS id, i % 100 AS val FROM range(122880) t(i); + +query I +SELECT count(*) FROM t_full WHERE val = 42; +---- +1229 + +statement ok +INSERT INTO t_full +SELECT i + 122880, (i + 122880) % 100 +FROM range(1000) t(i); + +query TT +EXPLAIN SELECT count(*) FROM t_full WHERE val = 42; +---- +physical_plan :.*UNGROUPED_AGGREGATE.*Aggregates:.*count_star\(\).*CACHE_RECORDER.*Table OID: [0-9]+.*Filter Key:.*\(val = CAST\(42 AS BIGINT\)\).*Row ID Column: 1.*SEQ_SCAN.*Table:.*memory\.main\.t_full.*Type: Sequential Scan.*Projections: val.*Filters:.*val=42.*__condition_cache_filter.*\(rowid\).*~1,265 rows.* + +# Truncating queries skip side-effect cache building on a miss. +query TT +EXPLAIN SELECT * FROM t WHERE val = 7 LIMIT 1; +---- +physical_plan :.*PROJECTION.*id.*val.*~0 rows.*STREAMING_LIMIT.*SEQ_SCAN.*Table: memory\.main\.t.*Type: Sequential Scan.*Projections:.*val.*id.*Filters: val=7.*~11 rows.* diff --git a/test/sql/condition_cache_incremental.test b/test/sql/condition_cache_incremental.test new file mode 100644 index 0000000..b5aca53 --- /dev/null +++ b/test/sql/condition_cache_incremental.test @@ -0,0 +1,94 @@ +# name: test/sql/condition_cache_incremental.test +# description: Optimizer injects recorder on miss; cache builds as a query side-effect. +# group: [sql] + +require query_condition_cache + +statement ok +CREATE TABLE t AS SELECT i AS id, i % 100 AS val FROM range(500000) t(i); + +query IIII +SELECT * FROM condition_cache_info('t', 'val = 42'); +---- +0 0 0 0 + +query I +SELECT count(*) FROM t WHERE val = 42; +---- +5000 + +query IIII +SELECT * FROM condition_cache_info('t', 'val = 42'); +---- +5 5 245 245 + +query I +SELECT count(*) FROM t WHERE val = 42; +---- +5000 + +query IIII +SELECT * FROM condition_cache_info('t', 'val = 42'); +---- +5 5 245 245 + +query I +SELECT count(*) FROM t WHERE id < 3000; +---- +3000 + +query IIII +SELECT * FROM condition_cache_info('t', 'id < 3000'); +---- +5 5 2 245 + +query I +SELECT count(*) FROM t WHERE id < 3000; +---- +3000 + +query IIII +SELECT * FROM condition_cache_info('t', 'id < 3000'); +---- +5 5 2 245 + +query I +SELECT count(*) FROM t WHERE id < 0; +---- +0 + +query IIII +SELECT * FROM condition_cache_info('t', 'id < 0'); +---- +0 0 0 0 + +query I +SELECT count(*) FROM t WHERE id < 0; +---- +0 + +statement ok +SET use_query_condition_cache = false; + +query I +SELECT count(*) FROM t WHERE val = 42; +---- +5000 + +query IIII +SELECT * FROM condition_cache_info('t', 'val = 42'); +---- +0 0 0 0 + +statement ok +SET use_query_condition_cache = true; + +query I +SELECT count(*) FROM t WHERE val = 42; +---- +5000 + +query IIII +SELECT * FROM condition_cache_info('t', 'val = 42'); +---- +5 5 245 245 diff --git a/test/sql/condition_cache_info.test b/test/sql/condition_cache_info.test index cfd630f..76a2d86 100644 --- a/test/sql/condition_cache_info.test +++ b/test/sql/condition_cache_info.test @@ -36,7 +36,7 @@ SELECT * FROM condition_cache_build('t', 'id < 3000'); query IIII SELECT * FROM condition_cache_info('t', 'id < 3000'); ---- -1 5 2 245 +5 5 2 245 # Qualified table name works query IIII diff --git a/test/unittest/CMakeLists.txt b/test/unittest/CMakeLists.txt index 1f9074d..ee3dbe2 100644 --- a/test/unittest/CMakeLists.txt +++ b/test/unittest/CMakeLists.txt @@ -17,6 +17,7 @@ set(QUERY_CACHE_UNITTEST_OBJECTS test_logical_cache_invalidator.cpp test_normalize_expression.cpp test_optimizer_invalidation.cpp + test_physical_cache_recorder.cpp test_physical_cache_invalidator.cpp) add_executable(unittest_query_condition_cache ${QUERY_CACHE_UNITTEST_OBJECTS}) diff --git a/test/unittest/test_bitvector.cpp b/test/unittest/test_bitvector.cpp index 4466a47..22a0ff3 100644 --- a/test/unittest/test_bitvector.cpp +++ b/test/unittest/test_bitvector.cpp @@ -56,4 +56,82 @@ TEST_CASE("RowGroupFilter - basic operations", "[bitvector]") { REQUIRE_FALSE(a.VectorHasRows(0)); } } + +TEST_CASE("RowGroupFilter - observed bitmask", "[bitvector]") { + SECTION("default constructor leaves observed bits all zero") { + RowGroupFilter bv; + for (idx_t i = 0; i < VECTORS_PER_ROW_GROUP; ++i) { + REQUIRE_FALSE(bv.IsObserved(i)); + } + REQUIRE_FALSE(bv.IsFullyObserved()); + } + + SECTION("vector-of-indices constructor leaves observed bits all zero") { + RowGroupFilter bv({0, 5}); + REQUIRE_FALSE(bv.IsObserved(0)); + REQUIRE_FALSE(bv.IsObserved(5)); + } + + SECTION("SetObserved sets the bit") { + RowGroupFilter bv; + bv.SetObserved(3); + bv.SetObserved(7); + REQUIRE(bv.IsObserved(3)); + REQUIRE(bv.IsObserved(7)); + REQUIRE_FALSE(bv.IsObserved(4)); + } + + SECTION("IsFullyObserved true only when every vec bit set") { + RowGroupFilter bv; + for (idx_t i = 0; i < VECTORS_PER_ROW_GROUP; ++i) { + bv.SetObserved(i); + } + REQUIRE(bv.IsFullyObserved()); + } + + SECTION("MarkFullyObserved sets every observed bit") { + RowGroupFilter bv; + bv.MarkFullyObserved(); + REQUIRE(bv.IsFullyObserved()); + for (idx_t i = 0; i < VECTORS_PER_ROW_GROUP; ++i) { + REQUIRE(bv.IsObserved(i)); + } + } + + SECTION("MergeFrom ORs observed bits") { + RowGroupFilter a; + a.SetObserved(1); + a.SetObserved(3); + RowGroupFilter b; + b.SetObserved(2); + b.SetObserved(3); + a.MergeFrom(b); + REQUIRE(a.IsObserved(1)); + REQUIRE(a.IsObserved(2)); + REQUIRE(a.IsObserved(3)); + REQUIRE_FALSE(a.IsObserved(0)); + } + + SECTION("MergeFrom from an empty filter preserves existing observed bits") { + RowGroupFilter a; + a.SetObserved(5); + RowGroupFilter empty; + a.MergeFrom(empty); + REQUIRE(a.IsObserved(5)); + } +} + +TEST_CASE("ConditionCacheEntry - unobserved vector ranges", "[bitvector]") { + ConditionCacheEntry entry; + entry.SetObservedVector(/*rg_idx=*/0, /*vec_idx=*/0); + entry.SetObservedVector(/*rg_idx=*/0, /*vec_idx=*/1); + for (idx_t rg = 1; rg < 5; ++rg) { + entry.MarkRowGroupFullyObserved(rg); + } + + auto ranges = entry.GetUnobservedVectorRanges(/*total_rows=*/500000); + REQUIRE(ranges.size() == 1); + REQUIRE(ranges[0].start_row == 2 * STANDARD_VECTOR_SIZE); + REQUIRE(ranges[0].count == DEFAULT_ROW_GROUP_SIZE - (2 * STANDARD_VECTOR_SIZE)); +} } // namespace duckdb diff --git a/test/unittest/test_build_cache_entry.cpp b/test/unittest/test_build_cache_entry.cpp index b7458f1..0e51d43 100644 --- a/test/unittest/test_build_cache_entry.cpp +++ b/test/unittest/test_build_cache_entry.cpp @@ -1,5 +1,6 @@ #include "catch/catch.hpp" #include "query_condition_cache_functions.hpp" +#include "query_condition_cache_state.hpp" #include "duckdb/catalog/catalog.hpp" #include "duckdb/catalog/catalog_entry/duck_table_entry.hpp" @@ -36,6 +37,9 @@ TEST_CASE("BuildCacheEntry - basic predicate", "[build_cache_entry]") { REQUIRE(entry != nullptr); REQUIRE(entry->RowGroupCount() == 5); + for (idx_t rg = 0; rg < 5; ++rg) { + REQUIRE(entry->GetObservedVectorCount(rg) == VECTORS_PER_ROW_GROUP); + } } SECTION("selective predicate") { @@ -56,6 +60,9 @@ TEST_CASE("BuildCacheEntry - basic predicate", "[build_cache_entry]") { REQUIRE(entry->RowGroupVectorHasQualifyingRows(0, 1)); REQUIRE_FALSE(entry->RowGroupVectorHasQualifyingRows(0, 2)); REQUIRE(entry->RowGroupIsCompletelyEmpty(1)); + for (idx_t rg = 0; rg < 5; ++rg) { + REQUIRE(entry->GetObservedVectorCount(rg) == VECTORS_PER_ROW_GROUP); + } } SECTION("odd values pass, even values don't") { @@ -72,6 +79,9 @@ TEST_CASE("BuildCacheEntry - basic predicate", "[build_cache_entry]") { REQUIRE(entry != nullptr); REQUIRE(entry->RowGroupCount() == 5); + for (idx_t rg = 0; rg < 5; ++rg) { + REQUIRE(entry->GetObservedVectorCount(rg) == VECTORS_PER_ROW_GROUP); + } } SECTION("no matching rows") { @@ -90,6 +100,9 @@ TEST_CASE("BuildCacheEntry - basic predicate", "[build_cache_entry]") { REQUIRE(entry->RowGroupCount() == 5); REQUIRE(entry->RowGroupIsCompletelyEmpty(0)); REQUIRE(entry->RowGroupIsCompletelyEmpty(4)); + for (idx_t rg = 0; rg < 5; ++rg) { + REQUIRE(entry->GetObservedVectorCount(rg) == VECTORS_PER_ROW_GROUP); + } } } } // namespace duckdb diff --git a/test/unittest/test_filter.cpp b/test/unittest/test_filter.cpp index be5f68c..fc64094 100644 --- a/test/unittest/test_filter.cpp +++ b/test/unittest/test_filter.cpp @@ -7,6 +7,7 @@ #include "duckdb/planner/expression/bound_function_expression.hpp" #include "duckdb/planner/filter/expression_filter.hpp" #include "duckdb/planner/expression/bound_reference_expression.hpp" +#include "duckdb/planner/operator/logical_extension_operator.hpp" #include "duckdb/planner/operator/logical_get.hpp" #include "duckdb/storage/statistics/numeric_stats.hpp" #include "test_helpers.hpp" @@ -28,6 +29,19 @@ optional_ptr FindLogicalGet(LogicalOperator &op) { return nullptr; } +bool ContainsCacheRecorder(LogicalOperator &op) { + if (op.type == LogicalOperatorType::LOGICAL_EXTENSION_OPERATOR && + op.Cast().GetExtensionName() == "query_condition_cache_recorder") { + return true; + } + for (auto &child : op.children) { + if (ContainsCacheRecorder(*child)) { + return true; + } + } + return false; +} + } // namespace TEST_CASE("CacheExpressionFilter - CheckStatistics", "[query_condition_cache]") { @@ -36,6 +50,7 @@ TEST_CASE("CacheExpressionFilter - CheckStatistics", "[query_condition_cache]") entry->SetQualifyingVector(/*rg_idx=*/0, /*vec_idx=*/5); entry->EnsureRowGroup(/*rg_idx=*/1); entry->SetQualifyingVector(/*rg_idx=*/2, /*vec_idx=*/10); + entry->MarkAllRowGroupsFullyObserved(); auto dummy_expr = make_uniq(LogicalType {LogicalTypeId::BIGINT}, 0); CacheExpressionFilter filter(std::move(dummy_expr), entry); @@ -129,4 +144,61 @@ TEST_CASE("Optimizer injects cache filter into LogicalGet", "[query_condition_ca REQUIRE(get->table_filters.filters.find(COLUMN_IDENTIFIER_ROW_ID) == get->table_filters.filters.end()); } } + +TEST_CASE("Recorder backfill completes vectors skipped by pushdown", "[query_condition_cache]") { + DuckDB db(nullptr); + Connection con(db); + + REQUIRE_NO_FAIL(con.Query("LOAD query_condition_cache")); + REQUIRE_NO_FAIL(con.Query("CREATE TABLE t AS SELECT i AS id, i % 100 AS val FROM range(500000) t(i)")); + + REQUIRE_NO_FAIL(con.Query("SELECT count(*) FROM t WHERE id < 3000")); + + auto plan = con.ExtractPlan("SELECT count(*) FROM t WHERE id < 3000"); + REQUIRE(plan != nullptr); + REQUIRE_FALSE(ContainsCacheRecorder(*plan)); +} + +TEST_CASE("CacheExpressionFilter - observed-bit gating", "[query_condition_cache]") { + SECTION("empty row group with no observed bits: no pruning") { + auto entry = make_shared_ptr(); + entry->EnsureRowGroup(/*rg_idx=*/1); + + auto dummy_expr = make_uniq(LogicalType {LogicalTypeId::BIGINT}, 0); + CacheExpressionFilter filter(std::move(dummy_expr), entry); + + auto stats = NumericStats::CreateUnknown(LogicalType {LogicalTypeId::BIGINT}); + NumericStats::SetMin(stats, Value::BIGINT(122880)); + NumericStats::SetMax(stats, Value::BIGINT(200000)); + REQUIRE(filter.CheckStatistics(stats) == FilterPropagateResult::NO_PRUNING_POSSIBLE); + } + + SECTION("empty row group becomes prunable once marked fully observed") { + auto entry = make_shared_ptr(); + entry->EnsureRowGroup(/*rg_idx=*/1); + entry->MarkAllRowGroupsFullyObserved(); + + auto dummy_expr = make_uniq(LogicalType {LogicalTypeId::BIGINT}, 0); + CacheExpressionFilter filter(std::move(dummy_expr), entry); + + auto stats = NumericStats::CreateUnknown(LogicalType {LogicalTypeId::BIGINT}); + NumericStats::SetMin(stats, Value::BIGINT(122880)); + NumericStats::SetMax(stats, Value::BIGINT(200000)); + REQUIRE(filter.CheckStatistics(stats) == FilterPropagateResult::FILTER_ALWAYS_FALSE); + } + + SECTION("range with one fully-observed empty rg and one absent rg: no pruning") { + auto entry = make_shared_ptr(); + entry->EnsureRowGroup(/*rg_idx=*/1); + entry->MarkAllRowGroupsFullyObserved(); + + auto dummy_expr = make_uniq(LogicalType {LogicalTypeId::BIGINT}, 0); + CacheExpressionFilter filter(std::move(dummy_expr), entry); + + auto stats = NumericStats::CreateUnknown(LogicalType {LogicalTypeId::BIGINT}); + NumericStats::SetMin(stats, Value::BIGINT(122880)); + NumericStats::SetMax(stats, Value::BIGINT(300000)); + REQUIRE(filter.CheckStatistics(stats) == FilterPropagateResult::NO_PRUNING_POSSIBLE); + } +} } // namespace duckdb diff --git a/test/unittest/test_physical_cache_recorder.cpp b/test/unittest/test_physical_cache_recorder.cpp new file mode 100644 index 0000000..d3376a4 --- /dev/null +++ b/test/unittest/test_physical_cache_recorder.cpp @@ -0,0 +1,127 @@ +#include "catch/catch.hpp" + +#include "logical_cache_recorder.hpp" +#include "physical_cache_recorder.hpp" +#include "query_condition_cache_state.hpp" + +namespace duckdb { + +namespace { + +struct RecorderObservation { + idx_t rg_idx; + idx_t vec_idx; + bool has_qualifying; +}; + +void RecordSequence(ConditionCacheEntry &local_entry, const vector &observations) { + for (const auto &obs : observations) { + PhysicalCacheRecorder::RecordChunkObservation(local_entry, obs.rg_idx, obs.vec_idx, obs.has_qualifying); + } +} + +} // namespace + +TEST_CASE("PhysicalCacheRecorder - marks observed bit and qualifying bit per chunk", "[physical_recorder]") { + ConditionCacheEntry local_entry; + + RecordSequence(local_entry, {{.rg_idx = 5, .vec_idx = 0, .has_qualifying = true}, + {.rg_idx = 5, .vec_idx = 1, .has_qualifying = false}, + {.rg_idx = 5, .vec_idx = 2, .has_qualifying = true}}); + + REQUIRE(local_entry.HasRowGroup(5)); + REQUIRE(local_entry.GetObservedVectorCount(5) == 3); + REQUIRE(local_entry.RowGroupVectorHasQualifyingRows(5, 0)); + REQUIRE_FALSE(local_entry.RowGroupVectorHasQualifyingRows(5, 1)); + REQUIRE(local_entry.RowGroupVectorHasQualifyingRows(5, 2)); +} + +TEST_CASE("PhysicalCacheRecorder - non-contiguous vecs set bits exactly where observed", "[physical_recorder]") { + ConditionCacheEntry local_entry; + + RecordSequence(local_entry, {{.rg_idx = 0, .vec_idx = 0, .has_qualifying = true}, + {.rg_idx = 0, .vec_idx = 3, .has_qualifying = false}, + {.rg_idx = 0, .vec_idx = 7, .has_qualifying = true}}); + + REQUIRE(local_entry.GetObservedVectorCount(0) == 3); + REQUIRE(local_entry.VectorPassesFilter(0, 0) == true); + REQUIRE(local_entry.VectorPassesFilter(0, 1) == true); + REQUIRE(local_entry.VectorPassesFilter(0, 3) == false); + REQUIRE(local_entry.VectorPassesFilter(0, 4) == true); + REQUIRE(local_entry.VectorPassesFilter(0, 7) == true); +} + +TEST_CASE("PhysicalCacheRecorder - single chunk keys rg and sets one bit", "[physical_recorder]") { + ConditionCacheEntry local_entry; + RecordSequence(local_entry, {{.rg_idx = 0, .vec_idx = 0, .has_qualifying = true}}); + + REQUIRE(local_entry.HasRowGroup(0)); + REQUIRE(local_entry.GetObservedVectorCount(0) == 1); + REQUIRE(local_entry.RowGroupVectorHasQualifyingRows(0, 0)); +} + +TEST_CASE("PhysicalCacheRecorder - rg with no qualifying rows is keyed with observed bits set", "[physical_recorder]") { + ConditionCacheEntry local_entry; + + RecordSequence(local_entry, {{.rg_idx = 2, .vec_idx = 0, .has_qualifying = false}, + {.rg_idx = 2, .vec_idx = 1, .has_qualifying = false}, + {.rg_idx = 2, .vec_idx = 2, .has_qualifying = false}}); + + REQUIRE(local_entry.HasRowGroup(2)); + REQUIRE(local_entry.GetObservedVectorCount(2) == 3); + REQUIRE(local_entry.RowGroupIsCompletelyEmpty(2)); +} + +TEST_CASE("PhysicalCacheRecorder - merging two task-local entries ORs observed and qualifying bits", + "[physical_recorder]") { + ConditionCacheEntry task_a; + RecordSequence(task_a, {{.rg_idx = 0, .vec_idx = 0, .has_qualifying = true}, + {.rg_idx = 0, .vec_idx = 2, .has_qualifying = true}, + {.rg_idx = 1, .vec_idx = 1, .has_qualifying = true}}); + + ConditionCacheEntry task_b; + RecordSequence(task_b, {{.rg_idx = 0, .vec_idx = 1, .has_qualifying = false}, + {.rg_idx = 2, .vec_idx = 0, .has_qualifying = true}}); + + auto destination = make_shared_ptr(); + destination->MergeFrom(task_a); + destination->MergeFrom(task_b); + + REQUIRE(destination->GetObservedVectorCount(0) == 3); + REQUIRE(destination->GetObservedVectorCount(1) == 1); + REQUIRE(destination->GetObservedVectorCount(2) == 1); + + REQUIRE(destination->RowGroupVectorHasQualifyingRows(0, 0)); + REQUIRE_FALSE(destination->RowGroupVectorHasQualifyingRows(0, 1)); + REQUIRE(destination->RowGroupVectorHasQualifyingRows(0, 2)); + REQUIRE(destination->RowGroupVectorHasQualifyingRows(1, 1)); + REQUIRE(destination->RowGroupVectorHasQualifyingRows(2, 0)); +} + +TEST_CASE("PhysicalCacheRecorder - second pass accumulates observations across queries", "[physical_recorder]") { + ConditionCacheEntry pass1; + RecordSequence(pass1, {{.rg_idx = 0, .vec_idx = 0, .has_qualifying = true}, + {.rg_idx = 0, .vec_idx = 1, .has_qualifying = false}}); + + ConditionCacheEntry pass2; + RecordSequence(pass2, {{.rg_idx = 0, .vec_idx = 2, .has_qualifying = true}, + {.rg_idx = 0, .vec_idx = 3, .has_qualifying = false}}); + + auto store = make_shared_ptr(); + store->MergeFrom(pass1); + REQUIRE(store->GetObservedVectorCount(0) == 2); + store->MergeFrom(pass2); + REQUIRE(store->GetObservedVectorCount(0) == 4); + + REQUIRE(store->RowGroupVectorHasQualifyingRows(0, 0)); + REQUIRE_FALSE(store->RowGroupVectorHasQualifyingRows(0, 1)); + REQUIRE(store->RowGroupVectorHasQualifyingRows(0, 2)); + REQUIRE_FALSE(store->RowGroupVectorHasQualifyingRows(0, 3)); +} + +TEST_CASE("CacheRecorderOperatorExtension - GetName matches recorder's GetExtensionName", "[physical_recorder]") { + CacheRecorderOperatorExtension ext; + REQUIRE(ext.GetName() == "query_condition_cache_recorder"); +} + +} // namespace duckdb