Backfill Unobserved Vectors

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

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<ConditionCacheEntry> cache_entry;
+	shared_ptr<ConditionCacheEntry> metadata_entry;
+
+	LogicalCacheRecorder(idx_t table_oid_p, string canonical_key_p, unique_ptr<Expression> 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<Expression> backfill_predicate_p = nullptr,
+	                     shared_ptr<ConditionCacheEntry> cache_entry_p = nullptr,
+	                     shared_ptr<ConditionCacheEntry> metadata_entry_p = nullptr);
+
+	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;
+};
+
+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,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<ConditionCacheEntry> local_entry;
+	ExpressionExecutor expr_executor;
+};
+
+struct CacheRecorderGlobalState : public GlobalOperatorState {
+	concurrency::mutex lock;
+	vector<shared_ptr<ConditionCacheEntry>> 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<Expression> bound_predicate_p, idx_t rowid_column_index_p, string table_catalog_p,
+	                      string table_schema_p, string table_name_p, unique_ptr<Expression> backfill_predicate_p,
+	                      shared_ptr<ConditionCacheEntry> cache_entry_p,
+	                      shared_ptr<ConditionCacheEntry> metadata_entry_p, vector<LogicalType> types,
+	                      idx_t estimated_cardinality);
+
+	idx_t table_oid;
+	string canonical_key;
+	string table_catalog;
+	string table_schema;
+	string table_name;
+	unique_ptr<Expression> bound_predicate;
+	unique_ptr<Expression> backfill_predicate;
+	idx_t rowid_column_index;
+	shared_ptr<ConditionCacheEntry> cache_entry;
+	shared_ptr<ConditionCacheEntry> metadata_entry;
+
+	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;
+
+	static void RecordChunkObservation(ConditionCacheEntry &local_entry, idx_t rg_idx, idx_t vec_idx,
+	                                   bool has_qualifying);
+};
+
+} // namespace duckdb

src/include/query_condition_cache_functions.hpp

@@ -16,6 +16,9 @@ class Expression;
 // Modifies bound_expr in-place (remaps column indices to scan positions).
 shared_ptr<ConditionCacheEntry> BuildCacheEntry(ClientContext &context, DuckTableEntry &table_entry,
                                                 Expression &bound_expr);
+shared_ptr<ConditionCacheEntry> BuildCacheEntryForRanges(ClientContext &context, DuckTableEntry &table_entry,
+                                                         Expression &bound_expr,
+                                                         const vector<CacheObservationRange> &ranges);
 
 TableFunction ConditionCacheBuildFunction();
 TableFunction ConditionCacheInfoFunction();

src/include/query_condition_cache_optimizer.hpp

@@ -4,55 +4,61 @@
 
 #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<Expression> predicate;
+	unique_ptr<Expression> backfill_predicate;
+	shared_ptr<ConditionCacheEntry> entry;
+	shared_ptr<ConditionCacheEntry> 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<idx_t, shared_ptr<ConditionCacheEntry>> cache_apply_pending;
+	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.
+	                            bool inside_truncating, CacheOptimizerQueryState &state);
 	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);
+	static void InjectCacheRecorder(unique_ptr<LogicalOperator> &plan, idx_t table_oid, string canonical_key,
+	                                string table_catalog, string table_schema, string table_name,
+	                                unique_ptr<Expression> predicate, unique_ptr<Expression> backfill_predicate,
+	                                const shared_ptr<ConditionCacheEntry> &entry,
+	                                const shared_ptr<ConditionCacheEntry> &metadata_entry);
+	static shared_ptr<ConditionCacheEntry> GetPrunedRowGroupsFromTableFilters(ClientContext &context,
+	                                                                          const LogicalGet &get);
+	static idx_t EnsureRowIdChunkIndex(LogicalGet &get);
 };
 
 } // 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/atomic.hpp"
 #include "duckdb/common/types/hash.hpp"
 #include "duckdb/common/unordered_map.hpp"
@@ -15,25 +16,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<uint64_t, N> 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<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; call SetObserved explicitly when a vec was scanned.
 	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);
+	void MarkFullyObserved();
+	bool IsObserved(idx_t vector_index) const;
+	bool IsFullyObserved() const;
 	void MergeFrom(const RowGroupFilter &other);
 };
 
@@ -56,10 +60,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() {
@@ -73,7 +82,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) ---
@@ -84,20 +93,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<CacheObservationRange> 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<idx_t, bool> EraseRowGroups(const unordered_set<idx_t> &row_group_indices);
+	pair<idx_t, bool> EraseRowGroupsStartingAt(idx_t first_row_group);
 
 private:
 	mutable concurrency::mutex lock;
@@ -168,6 +187,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<idx_t> &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);
@@ -177,6 +198,7 @@ class ConditionCacheStore : public ObjectCacheEntry {
 
 	// Get or create the store from a client context
 	static shared_ptr<ConditionCacheStore> GetOrCreate(ClientContext &context);
+	static shared_ptr<ConditionCacheStore> GetOrCreate(DatabaseInstance &db);
 
 	// Record an optimizer lookup attempt.
 	void RecordAccess(bool hit);