diff --git a/CMakeLists.txt b/CMakeLists.txt index bfabf4f..2a32b73 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -22,7 +22,9 @@ include_directories(src/include) set(EXTENSION_SOURCES src/cache_invalidation_optimizer.cpp src/logical_cache_invalidator.cpp + src/logical_cache_recorder.cpp src/physical_cache_invalidator.cpp + src/physical_cache_recorder.cpp src/predicate_key_utils.cpp src/query_condition_cache_extension.cpp src/query_condition_cache_filter.cpp diff --git a/src/include/logical_cache_recorder.hpp b/src/include/logical_cache_recorder.hpp new file mode 100644 index 0000000..91eb59a --- /dev/null +++ b/src/include/logical_cache_recorder.hpp @@ -0,0 +1,36 @@ +#pragma once + +#include "duckdb/planner/operator/logical_extension_operator.hpp" +#include "physical_cache_recorder.hpp" + +namespace duckdb { + +// Logical wrapper for PhysicalCacheRecorder. Carries the bound predicate already +// remapped to chunk column positions and the canonical key for the store. +struct LogicalCacheRecorder : public LogicalExtensionOperator { + idx_t table_oid; + string canonical_key; + idx_t rowid_column_index; + + LogicalCacheRecorder(idx_t table_oid_p, string canonical_key_p, unique_ptr bound_predicate_p, + idx_t rowid_column_index_p); + + PhysicalOperator &CreatePlan(ClientContext &context, PhysicalPlanGenerator &planner) override; + vector GetColumnBindings() override; + string GetExtensionName() const override; + void Serialize(Serializer &serializer) const override; + +protected: + void ResolveTypes() override; +}; + +// Uses extension name "query_condition_cache_recorder" to avoid colliding with +// CacheInvalidatorOperatorExtension's "query_condition_cache". +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..bf01d5d --- /dev/null +++ b/src/include/physical_cache_recorder.hpp @@ -0,0 +1,58 @@ +#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); + + // Co-owned with the global state so bitvectors outlive this OperatorState. + shared_ptr local_entry; + ExpressionExecutor expr_executor; +}; + +struct CacheRecorderGlobalState : public GlobalOperatorState { + concurrency::mutex lock; + vector> task_local_entries DUCKDB_GUARDED_BY(lock); +}; + +// Pass-through operator injected above LogicalGet on cache miss. Observes scan output, +// evaluates the predicate per chunk, and merges thread-local bitvectors into the store +// at OperatorFinalize. The last vec each task observes is left uncommitted because it +// may be a partial tail. +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, + vector types, idx_t estimated_cardinality); + + idx_t table_oid; + string canonical_key; + unique_ptr bound_predicate; + idx_t rowid_column_index; + + 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; + + // Public so unit tests can drive the algorithm with synthetic (rg, vec, qualifying) tuples + // without a real pipeline. + 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_optimizer.hpp b/src/include/query_condition_cache_optimizer.hpp index 221646a..cf03e1a 100644 --- a/src/include/query_condition_cache_optimizer.hpp +++ b/src/include/query_condition_cache_optimizer.hpp @@ -10,17 +10,25 @@ 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 canonical_key; + // BoundColumnRefs already remapped to chunk positions for the recorder. + unique_ptr predicate; +}; + +// Query-scoped state passed from pre-optimize to post-optimize. 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. + // table_index -> entry to bind the filter against. unordered_map> cache_apply_pending; + // table_index -> recorder injection info, populated on cache miss only. + unordered_map cache_recorder_pending; void QueryEnd(ClientContext &context, optional_ptr error) override { cache_apply_pending.clear(); + cache_recorder_pending.clear(); } }; @@ -28,31 +36,25 @@ 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. 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); + + // Wraps `plan` (a LogicalGet) with a LogicalCacheRecorder; reassigns `plan` so the + // parent's children pointer updates transparently. + static void InjectCacheRecorder(ClientContext &context, unique_ptr &plan, + RecorderInjectionInfo &&info); }; } // namespace duckdb diff --git a/src/include/query_condition_cache_state.hpp b/src/include/query_condition_cache_state.hpp index 5be7bba..9e5ba40 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,29 @@ 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 observed to have at least one qualifying row. +// observed[i] = 1 iff vec i has been observed. An unobserved vec is +// "unknown" and must be scanned. 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; set bits explicitly if you need pruning semantics. 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); + bool IsObserved(idx_t vector_index) const; + bool IsFullyObserved() const; + // OR-merge of both bitsets. void MergeFrom(const RowGroupFilter &other); }; @@ -77,22 +82,27 @@ struct ConditionCacheEntry : public ObjectCacheEntry { // --- Thread-safe API (each method acquires `lock` internally) --- - // Ensure a row group key exists (empty filter). Used when recording fully excluded row groups. + // Ensure a row group key exists (empty matching, no observed bits set). void EnsureRowGroup(idx_t rg_idx); - // Mark that vector `vec_idx` within row group `rg_idx` has at least one qualifying row. 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 manual full-table builds to declare every keyed rg fully observed. + void MarkAllRowGroupsFullyObserved(); + // Mark a single vec as observed. Safe per-task because store-side MergeFrom ORs observed. + 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). + // False only if rg is cached AND vec is observed AND matching bit = 0. 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 rg in range is cached, fully observed, and empty. bool StatisticsRangeIsAllEmptyCached(idx_t min_rg, idx_t max_rg) const; idx_t RowGroupCount() const; bool HasRowGroup(idx_t rg_idx) const; + // Popcount of the rg's observed bitmask; 0 if rg is not present. + idx_t GetObservedVectorCount(idx_t rg_idx) const; + // Raw matching bit, ignores observed. Test-only; production uses VectorPassesFilter. 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). + // Raw emptiness of matching bits, ignores observed. Test-only. bool RowGroupIsCompletelyEmpty(idx_t rg_idx) const; // Erase row group keys; returns (number of keys removed, whether the map is now empty). diff --git a/src/logical_cache_recorder.cpp b/src/logical_cache_recorder.cpp new file mode 100644 index 0000000..8b81b46 --- /dev/null +++ b/src/logical_cache_recorder.cpp @@ -0,0 +1,78 @@ +#include "logical_cache_recorder.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_reference_expression.hpp" + +namespace duckdb { + +LogicalCacheRecorder::LogicalCacheRecorder(idx_t table_oid_p, string canonical_key_p, + unique_ptr bound_predicate_p, idx_t rowid_column_index_p) + : table_oid(table_oid_p), canonical_key(std::move(canonical_key_p)), rowid_column_index(rowid_column_index_p) { + expressions.push_back(std::move(bound_predicate_p)); +} + +PhysicalOperator &LogicalCacheRecorder::CreatePlan(ClientContext &context, PhysicalPlanGenerator &planner) { + auto &child_plan = planner.CreatePlan(*children[0]); + auto bound_predicate = std::move(expressions[0]); + auto &op = planner.Make(table_oid, canonical_key, std::move(bound_predicate), + rowid_column_index, 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); +} + +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"); + + unique_ptr bound_predicate; + if (!exprs.empty()) { + bound_predicate = std::move(exprs[0]); + } else { + // Placeholder keeps the operator structurally intact for verification round-trips. + bound_predicate = make_uniq(LogicalType {LogicalTypeId::BOOLEAN}, 0); + } + return make_uniq(oid, std::move(key), std::move(bound_predicate), rowid_col); +} + +} // namespace duckdb diff --git a/src/physical_cache_recorder.cpp b/src/physical_cache_recorder.cpp new file mode 100644 index 0000000..23b4878 --- /dev/null +++ b/src/physical_cache_recorder.cpp @@ -0,0 +1,144 @@ +#include "physical_cache_recorder.hpp" + +#include "concurrency/annotated_lock.hpp" + +#include "duckdb/common/assert.hpp" +#include "duckdb/common/numeric_utils.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, + 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)), bound_predicate(std::move(bound_predicate_p)), + rowid_column_index(rowid_column_index_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 { + +// Register so the entry outlives this OperatorState (destroyed before OperatorFinalize). +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); +} + +} // 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); + + auto first_idx = rowid_data.sel->get_index(0); + if (!rowid_data.validity.RowIsValid(first_idx)) { + return OperatorResultType::NEED_MORE_INPUT; + } + auto first_row_id = NumericCast(rowids[first_idx]); + if (first_row_id >= NumericCast(MAX_ROW_ID)) { + // Transaction-local storage rows have no stable cache identity. + return OperatorResultType::NEED_MORE_INPUT; + } + + idx_t rg_idx = first_row_id / DEFAULT_ROW_GROUP_SIZE; + idx_t vec_idx = (first_row_id % DEFAULT_ROW_GROUP_SIZE) / STANDARD_VECTOR_SIZE; + + SelectionVector sel(input.size()); + idx_t match_count = local_state.expr_executor.SelectExpression(input, sel); + + RecordChunkObservation(*local_state.local_entry, rg_idx, vec_idx, /*has_qualifying=*/match_count > 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); + } +} + +OperatorFinalResultType PhysicalCacheRecorder::OperatorFinalize(Pipeline &pipeline, Event &event, + ClientContext &context, + OperatorFinalizeInput &input) const { + auto &global_state = input.global_state.Cast(); + + // TODO: backfill rgs scan fully skipped (zone-map or column-filter prune) by consulting + // storage for the rg count and keying missing entries with bit=0 + watermark=FULL. Only + // safe when scan ran to completion; pair with LogicalLimit detection in the optimizer. + auto store = ConditionCacheStore::GetOrCreate(context); + CacheKey key {table_oid, canonical_key}; + auto destination = store->Lookup(context, key); + if (!destination) { + destination = make_shared_ptr(); + } + + { + concurrency::lock_guard guard(global_state.lock); + for (const auto &task_entry : global_state.task_local_entries) { + destination->MergeFrom(*task_entry); + } + } + + store->Upsert(context, key, std::move(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..09630d9 100644 --- a/src/query_condition_cache_extension.cpp +++ b/src/query_condition_cache_extension.cpp @@ -8,6 +8,7 @@ #include "duckdb/optimizer/optimizer_extension.hpp" #include "cache_invalidation_optimizer.hpp" #include "logical_cache_invalidator.hpp" +#include "logical_cache_recorder.hpp" #include "query_condition_cache_filter.hpp" #include "query_condition_cache_functions.hpp" #include "query_condition_cache_optimizer.hpp" @@ -43,6 +44,7 @@ void LoadInternal(ExtensionLoader &loader) { 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..fb46e56 100644 --- a/src/query_condition_cache_functions.cpp +++ b/src/query_condition_cache_functions.cpp @@ -254,6 +254,8 @@ shared_ptr BuildCacheEntry(ClientContext &context, DuckTabl auto entry = make_shared_ptr(); MergeLocalCacheEntries(local_entries, entry); + // Full-table build observes every rg end-to-end; trust bits for pruning. + entry->MarkAllRowGroupsFullyObserved(); return entry; } diff --git a/src/query_condition_cache_optimizer.cpp b/src/query_condition_cache_optimizer.cpp index d531da1..9dac388 100644 --- a/src/query_condition_cache_optimizer.cpp +++ b/src/query_condition_cache_optimizer.cpp @@ -1,5 +1,6 @@ #include "query_condition_cache_optimizer.hpp" +#include "logical_cache_recorder.hpp" #include "query_condition_cache_filter.hpp" #include "predicate_key_utils.hpp" #include "query_condition_cache_functions.hpp" @@ -42,14 +43,32 @@ 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); } catch (...) { // Defense in depth: skip cache optimization rather than failing the query. query_state->cache_apply_pending.clear(); + query_state->cache_recorder_pending.clear(); } } +namespace { + +// BoundColumnRef.column_index == chunk position at runtime, so we can rewrite directly +// to BoundReference(column_index). +void ConvertColumnRefsToChunkRefs(unique_ptr &expr) { + if (expr->GetExpressionClass() == ExpressionClass::BOUND_COLUMN_REF) { + auto &colref = expr->Cast(); + expr = make_uniq(colref.alias, colref.return_type, colref.binding.column_index); + return; + } + ExpressionIterator::EnumerateChildren(*expr, + [&](unique_ptr &child) { ConvertColumnRefsToChunkRefs(child); }); +} + +} // namespace + void QueryConditionCacheOptimizer::PreOptimizeWalk(ClientContext &context, unique_ptr &plan, bool inside_dml, CacheOptimizerQueryState &state) { // Skip cache building inside DML subplans @@ -103,62 +122,30 @@ void QueryConditionCacheOptimizer::PreOptimizeWalk(ClientContext &context, uniqu auto entry = store->Lookup(context, key); 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); + // Upsert the empty entry up front so the cache filter's bind_data and the recorder's + // Finalize Lookup resolve to the same shared_ptr. + entry = make_shared_ptr(); + store->Upsert(context, key, entry); + + // TODO: Also inject the recorder on a partial cache hit so the watermark can + // advance on later queries and DML-dropped rgs get re-observed. Must skip under + // LogicalLimit: a truncated scan cannot distinguish "unobserved" from "no match". + vector> cloned; + cloned.reserve(filter.expressions.size()); + for (const auto &expr : filter.expressions) { + auto copy = expr->Copy(); + ConvertColumnRefsToChunkRefs(copy); + cloned.push_back(std::move(copy)); } - } - - 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 + auto predicate = CombineWithAnd(std::move(cloned)); + predicate = + BoundCastExpression::AddCastToType(context, std::move(predicate), LogicalType {LogicalTypeId::BOOLEAN}); -shared_ptr QueryConditionCacheOptimizer::BuildCacheForPredicate( - ClientContext &context, const vector> &expressions, LogicalGet &get) { - auto table_ptr = get.GetTable(); - if (!table_ptr) { - return nullptr; + state.cache_recorder_pending[get.table_index] = + RecorderInjectionInfo {table->oid, key.filter_key, std::move(predicate)}; } - 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)); - } - - 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[get.table_index] = std::move(entry); } void QueryConditionCacheOptimizer::PostOptimizeWalk(ClientContext &context, unique_ptr &plan, @@ -179,6 +166,13 @@ void QueryConditionCacheOptimizer::PostOptimizeWalk(ClientContext &context, uniq 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()) { + auto info = std::move(recorder_it->second); + state.cache_recorder_pending.erase(recorder_it); + InjectCacheRecorder(context, plan, std::move(info)); + } } void QueryConditionCacheOptimizer::InjectCacheFilter(ClientContext &context, LogicalGet &get, @@ -212,13 +206,57 @@ void QueryConditionCacheOptimizer::InjectCacheFilter(ClientContext &context, Log make_uniq(std::move(filter_expr), entry)); } +void QueryConditionCacheOptimizer::InjectCacheRecorder(ClientContext &context, unique_ptr &plan, + RecorderInjectionInfo &&info) { + D_ASSERT(plan->type == LogicalOperatorType::LOGICAL_GET); + auto &get = plan->Cast(); + auto &column_ids = get.GetMutableColumnIds(); + + // ROW_ID was added to column_ids by InjectCacheFilter; we also surface it through + // projection_ids so it reaches the recorder's input chunk. + idx_t rowid_column_ids_pos = column_ids.size(); + for (idx_t ii = 0; ii < column_ids.size(); ++ii) { + if (column_ids[ii].IsRowIdColumn()) { + rowid_column_ids_pos = ii; + break; + } + } + D_ASSERT(rowid_column_ids_pos < column_ids.size()); + + // Chunk-level position = column_ids position if no projection, otherwise the matching + // (or newly appended) projection_ids slot. + idx_t rowid_chunk_idx; + if (get.projection_ids.empty()) { + rowid_chunk_idx = rowid_column_ids_pos; + } else { + idx_t found = get.projection_ids.size(); + for (idx_t ii = 0; ii < get.projection_ids.size(); ++ii) { + if (get.projection_ids[ii] == rowid_column_ids_pos) { + found = ii; + break; + } + } + if (found < get.projection_ids.size()) { + rowid_chunk_idx = found; + } else { + get.projection_ids.push_back(rowid_column_ids_pos); + rowid_chunk_idx = get.projection_ids.size() - 1; + } + } + + auto recorder = make_uniq(info.table_oid, std::move(info.canonical_key), + std::move(info.predicate), rowid_chunk_idx); + recorder->children.push_back(std::move(plan)); + plan = std::move(recorder); +} + void QueryConditionCacheOptimizer::OptimizeFunction(OptimizerExtensionInput &input, unique_ptr &plan) { if (!IsSettingEnabled(input.context)) { return; } auto query_state = input.context.registered_state->Get(CacheOptimizerQueryState::NAME); - if (!query_state || query_state->cache_apply_pending.empty()) { + if (!query_state || (query_state->cache_apply_pending.empty() && query_state->cache_recorder_pending.empty())) { return; } diff --git a/src/query_condition_cache_state.cpp b/src/query_condition_cache_state.cpp index 1844ba5..c6e2c16 100644 --- a/src/query_condition_cache_state.cpp +++ b/src/query_condition_cache_state.cpp @@ -10,31 +10,37 @@ 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); +} + +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 ------- @@ -56,11 +62,7 @@ CacheEntryStats ConditionCacheEntry::ComputeStats(idx_t total_rows) const { 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 full_row_groups = total_rows / DEFAULT_ROW_GROUP_SIZE; @@ -107,12 +109,27 @@ void ConditionCacheEntry::MergeFrom(const ConditionCacheEntry &other) { } } +void ConditionCacheEntry::MarkAllRowGroupsFullyObserved() { + concurrency::lock_guard guard(lock); + for (auto &[rg_idx, filter] : bitvectors) { + filter.observed.set(); + } +} + +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,6 +140,9 @@ 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; } @@ -137,6 +157,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); diff --git a/test/sql/condition_cache_incremental.test b/test/sql/condition_cache_incremental.test new file mode 100644 index 0000000..9f90b87 --- /dev/null +++ b/test/sql/condition_cache_incremental.test @@ -0,0 +1,95 @@ +# 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); + +# Empty cache. +query IIII +SELECT * FROM condition_cache_info('t', 'val = 42'); +---- +0 0 0 0 + +# First query populates cache via recorder side-effect. +query I +SELECT count(*) FROM t WHERE val = 42; +---- +5000 + +query IIII +SELECT * FROM condition_cache_info('t', 'val = 42'); +---- +5 5 245 245 + +# Repeat: hit, no re-injection. +query I +SELECT count(*) FROM t WHERE val = 42; +---- +5000 + +query IIII +SELECT * FROM condition_cache_info('t', 'val = 42'); +---- +5 5 245 245 + +# Selective predicate: only rg 0 has matches; rgs 1..4 keyed as empty for pruning. +query I +SELECT count(*) FROM t WHERE id < 3000; +---- +3000 + +query IIII +SELECT * FROM condition_cache_info('t', 'id < 3000'); +---- +1 5 2 245 + +query I +SELECT count(*) FROM t WHERE id < 3000; +---- +3000 + +# Predicate matching nothing still keys every rg as observed empty. +query I +SELECT count(*) FROM t WHERE id < 0; +---- +0 + +query IIII +SELECT * FROM condition_cache_info('t', 'id < 0'); +---- +0 5 0 245 + +query I +SELECT count(*) FROM t WHERE id < 0; +---- +0 + +# Disable clears cache; re-enable starts from a clean slate. +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/unittest/CMakeLists.txt b/test/unittest/CMakeLists.txt index 1f9074d..ee3d206 100644 --- a/test/unittest/CMakeLists.txt +++ b/test/unittest/CMakeLists.txt @@ -17,7 +17,8 @@ set(QUERY_CACHE_UNITTEST_OBJECTS test_logical_cache_invalidator.cpp test_normalize_expression.cpp test_optimizer_invalidation.cpp - test_physical_cache_invalidator.cpp) + test_physical_cache_invalidator.cpp + test_physical_cache_recorder.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..8803a19 100644 --- a/test/unittest/test_bitvector.cpp +++ b/test/unittest/test_bitvector.cpp @@ -56,4 +56,59 @@ 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("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)); + } +} } // namespace duckdb diff --git a/test/unittest/test_build_cache_entry.cpp b/test/unittest/test_build_cache_entry.cpp index b7458f1..f63ff3b 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,11 @@ TEST_CASE("BuildCacheEntry - basic predicate", "[build_cache_entry]") { REQUIRE(entry != nullptr); REQUIRE(entry->RowGroupCount() == 5); + // BuildCacheEntry must finalise the watermark so VectorPassesFilter / CheckStatistics + // trust the bitvectors for pruning. + for (idx_t rg = 0; rg < 5; ++rg) { + REQUIRE(entry->GetObservedVectorCount(rg) == VECTORS_PER_ROW_GROUP); + } } SECTION("selective predicate") { @@ -56,6 +62,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 +81,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 +102,11 @@ TEST_CASE("BuildCacheEntry - basic predicate", "[build_cache_entry]") { REQUIRE(entry->RowGroupCount() == 5); REQUIRE(entry->RowGroupIsCompletelyEmpty(0)); REQUIRE(entry->RowGroupIsCompletelyEmpty(4)); + // Even when no rows match, every observed row group must be marked fully observed so + // CheckStatistics can prune them. + 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..80228bf 100644 --- a/test/unittest/test_filter.cpp +++ b/test/unittest/test_filter.cpp @@ -36,6 +36,9 @@ 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); + // CheckStatistics requires every row group in range to be fully observed before it can + // confidently report FILTER_ALWAYS_FALSE; emulate the full-build state for these tests. + entry->MarkAllRowGroupsFullyObserved(); auto dummy_expr = make_uniq(LogicalType {LogicalTypeId::BIGINT}, 0); CacheExpressionFilter filter(std::move(dummy_expr), entry); @@ -67,6 +70,51 @@ TEST_CASE("CacheExpressionFilter - CheckStatistics", "[query_condition_cache]") } } +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); // observed bits all zero + + 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(); + // rg 2 is intentionally absent: a range spanning rg 1 and rg 2 cannot be pruned because + // the cache has no information about rg 2. + + 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); + } +} + TEST_CASE("Optimizer injects cache filter into LogicalGet", "[query_condition_cache]") { DuckDB db(nullptr); Connection con(db); diff --git a/test/unittest/test_physical_cache_recorder.cpp b/test/unittest/test_physical_cache_recorder.cpp new file mode 100644 index 0000000..e1e34c7 --- /dev/null +++ b/test/unittest/test_physical_cache_recorder.cpp @@ -0,0 +1,131 @@ +#include "catch/catch.hpp" + +#include "logical_cache_recorder.hpp" +#include "physical_cache_recorder.hpp" +#include "query_condition_cache_state.hpp" + +namespace duckdb { + +namespace { + +// Two idx_t fields — designated initializers prevent arg-swap mistakes. +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; + + // Simulating column-filter-induced gaps: scan emits chunks only for vecs 0, 3, 7. + 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); + // Observed bits only where chunks arrived; intermediate vecs are NOT claimed observed. + REQUIRE(local_entry.VectorPassesFilter(0, 0) == true); // observed + qualifying -> pass + REQUIRE(local_entry.VectorPassesFilter(0, 1) == true); // unobserved -> pass-through + REQUIRE(local_entry.VectorPassesFilter(0, 3) == false); // observed + non-qualifying -> prune + REQUIRE(local_entry.VectorPassesFilter(0, 4) == true); // unobserved + REQUIRE(local_entry.VectorPassesFilter(0, 7) == true); // observed + qualifying -> pass +} + +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); + + // observed bits: union of tasks' observations. + REQUIRE(destination->GetObservedVectorCount(0) == 3); // vecs 0, 1, 2 + REQUIRE(destination->GetObservedVectorCount(1) == 1); // vec 1 + REQUIRE(destination->GetObservedVectorCount(2) == 1); // vec 0 + + 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