Replace sync cache build with recorder operator

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

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<Expression> bound_predicate_p,
+	                     idx_t rowid_column_index_p);
+
+	PhysicalOperator &CreatePlan(ClientContext &context, PhysicalPlanGenerator &planner) override;
+	vector<ColumnBinding> 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<LogicalExtensionOperator> Deserialize(Deserializer &deserializer) override;
+};
+
+} // namespace duckdb

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<ConditionCacheEntry> local_entry;
+	ExpressionExecutor expr_executor;
+};
+
+struct CacheRecorderGlobalState : public GlobalOperatorState {
+	concurrency::mutex lock;
+	vector<shared_ptr<ConditionCacheEntry>> 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<Expression> bound_predicate_p, idx_t rowid_column_index_p,
+	                      vector<LogicalType> types, idx_t estimated_cardinality);
+
+	idx_t table_oid;
+	string canonical_key;
+	unique_ptr<Expression> bound_predicate;
+	idx_t rowid_column_index;
+
+	unique_ptr<GlobalOperatorState> GetGlobalOperatorState(ClientContext &context) const override;
+	unique_ptr<OperatorState> 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<string> 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

src/include/query_condition_cache_optimizer.hpp

@@ -10,49 +10,51 @@ 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<Expression> 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<idx_t, shared_ptr<ConditionCacheEntry>> cache_apply_pending;
+	// table_index -> recorder injection info, populated on cache miss only.
+	unordered_map<idx_t, RecorderInjectionInfo> cache_recorder_pending;
 
 	void QueryEnd(ClientContext &context, optional_ptr<ErrorData> error) override {
 		cache_apply_pending.clear();
+		cache_recorder_pending.clear();
 	}
 };
 
 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<LogicalOperator> &plan);
-
-	// Post-optimize: inject cache filters into LogicalGet nodes that were matched pre-optimize.
 	static void OptimizeFunction(OptimizerExtensionInput &input, unique_ptr<LogicalOperator> &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<LogicalOperator> &plan, bool inside_dml,
 	                            CacheOptimizerQueryState &state);
 
-	// Build cache entry for a predicate on a table
-	static shared_ptr<ConditionCacheEntry>
-	BuildCacheForPredicate(ClientContext &context, const vector<unique_ptr<Expression>> &expressions, LogicalGet &get);
-
-	// Walk plan after built-in optimization and inject cache filters into matching table scans.
 	static void PostOptimizeWalk(ClientContext &context, unique_ptr<LogicalOperator> &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<ConditionCacheEntry> &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<LogicalOperator> &plan,
+	                                RecorderInjectionInfo &&info);
 };
 
 } // namespace duckdb

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 <bitset>
+
 #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<uint64_t, N> 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<uint64_t, BITVECTOR_ARRAY_SIZE> matching_vectors = {};
+	std::bitset<VECTORS_PER_ROW_GROUP> matching_vectors;
+	std::bitset<VECTORS_PER_ROW_GROUP> 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<idx_t> &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).

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<Expression> 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<PhysicalCacheRecorder>(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<ColumnBinding> 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<LogicalExtensionOperator> CacheRecorderOperatorExtension::Deserialize(Deserializer &deserializer) {
+	auto oid = deserializer.ReadProperty<idx_t>(400, "table_oid");
+	auto key = deserializer.ReadProperty<string>(401, "canonical_key");
+	auto rowid_col = deserializer.ReadProperty<idx_t>(402, "rowid_column_index");
+	auto exprs = deserializer.ReadPropertyWithDefault<vector<unique_ptr<Expression>>>(403, "expressions");
+
+	unique_ptr<Expression> 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<BoundReferenceExpression>(LogicalType {LogicalTypeId::BOOLEAN}, 0);
+	}
+	return make_uniq<LogicalCacheRecorder>(oid, std::move(key), std::move(bound_predicate), rowid_col);
+}
+
+} // namespace duckdb