Dev/defer bf init

extension_config.cmake

@@ -9,3 +9,7 @@ duckdb_extension_load(robust
 
 # Any extra extensions that should be built
 # e.g.: duckdb_extension_load(json)
+
+# tpch: statically linked so the benchmark_runner can CALL dbgen / run the
+# tpch benchmark suites (scripts/bench_tpch.sh). Required by `require tpch`.
+duckdb_extension_load(tpch)

scripts/bench_tpch.sh

@@ -24,6 +24,12 @@ set -e
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
 cd "$PROJECT_ROOT"
+# provision tpchdata/queries + answers from the submodule if missing
+if [ -z "$(ls -A "$PROJECT_ROOT/tpchdata/queries"/q*.sql 2>/dev/null)" ]; then
+    echo "TPCH queries not found; running setup_tpch_data.sh..."
+    "$SCRIPT_DIR/setup_tpch_data.sh"
+fi
+
 RUNNER="$PROJECT_ROOT/build/release/benchmark/benchmark_runner"
 PATTERN="q(02|03|07|08|10|11|17|18|21)\.benchmark"
 RUN_BASELINE=true

scripts/setup_tpch_data.sh

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+#
+# setup_tpch_data.sh — one-time setup for the TPCH benchmark queries and answers.
+#
+# Populates tpchdata/queries/ and tpchdata/answers/sf1/ from the canonical
+# copies shipped in the pinned duckdb submodule (duckdb/extension/tpch/dbgen/).
+# These are the same queries/answers DuckDB's own benchmark_runner uses, so they
+# stay version-locked to the submodule SHA. The tpch_sf1.duckdb database itself
+# is generated on demand by the benchmark_runner via CALL dbgen(sf=1).
+#
+# Usage:
+#   ./scripts/setup_tpch_data.sh           # skips if tpchdata/queries already populated
+#   ./scripts/setup_tpch_data.sh --force   # recopy even if present
+#
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+SRC_QUERIES="$PROJECT_ROOT/duckdb/extension/tpch/dbgen/queries"
+SRC_ANSWERS="$PROJECT_ROOT/duckdb/extension/tpch/dbgen/answers/sf1"
+DST_QUERIES="$PROJECT_ROOT/tpchdata/queries"
+DST_ANSWERS="$PROJECT_ROOT/tpchdata/answers/sf1"
+
+FORCE=0
+for arg in "$@"; do
+    case "$arg" in
+        --force|-f) FORCE=1 ;;
+        --help|-h)
+            sed -n '3,13p' "$0" | sed 's/^# \{0,1\}//'
+            exit 0
+            ;;
+        *)
+            echo "unknown arg: $arg (use --help)" >&2
+            exit 2
+            ;;
+    esac
+done
+
+if [ ! -d "$SRC_QUERIES" ] || [ ! -d "$SRC_ANSWERS" ]; then
+    echo "error: TPCH source files not found in the duckdb submodule" >&2
+    echo "  expected: $SRC_QUERIES" >&2
+    echo "        and: $SRC_ANSWERS" >&2
+    echo "  did you 'git submodule update --init --recursive'?" >&2
+    exit 1
+fi
+
+if [ -d "$DST_QUERIES" ] && [ -n "$(ls -A "$DST_QUERIES"/q*.sql 2>/dev/null)" ] && [ "$FORCE" -eq 0 ]; then
+    echo "$DST_QUERIES already populated; nothing to do (pass --force to recopy)."
+    exit 0
+fi
+
+mkdir -p "$DST_QUERIES" "$DST_ANSWERS"
+cp "$SRC_QUERIES"/q*.sql "$DST_QUERIES"/
+cp "$SRC_ANSWERS"/q*.csv "$DST_ANSWERS"/
+
+echo "Done."
+echo "  queries: $(ls "$DST_QUERIES"/q*.sql | wc -l | tr -d ' ') -> $DST_QUERIES"
+echo "  answers: $(ls "$DST_ANSWERS"/q*.csv | wc -l | tr -d ' ') -> $DST_ANSWERS"

src/operators/physical_create_filter.cpp

@@ -2,6 +2,9 @@
 #include "bloom_filter.hpp"
 #include "duckdb/execution/expression_executor.hpp"
 #include "duckdb/parallel/pipeline.hpp"
+#include "duckdb/parallel/base_pipeline_event.hpp"
+#include "duckdb/parallel/executor_task.hpp"
+#include "duckdb/parallel/task_scheduler.hpp"
 #include "utils/debug_utils.hpp"
 #include "robust_profiling.hpp"
 #include "probe_empty_registry.hpp"
@@ -189,12 +192,6 @@ static void UpdateMinMax(Vector &vec, idx_t count, ColumnMinMax &mm) {
 CreateFilterGlobalSinkState::CreateFilterGlobalSinkState(ClientContext &context, const PhysicalCreateFilter &op)
     : op(op) {
 	total_data = make_uniq<ColumnDataCollection>(context, op.types);
-	// initialize bloom filters upfront so Sink can insert directly
-	for (auto &entry : op.bloom_filter_map) {
-		if (entry.second) {
-			entry.second->Initialize(context, op.estimated_cardinality);
-		}
-	}
 	// resolve shared probe-empty flag once (single-threaded); forward pass only
 	if (op.is_forward_pass) {
 		auto reg = GetProbeEmptyRegistry(context);
@@ -239,22 +236,13 @@ SinkResultType PhysicalCreateFilter::Sink(ExecutionContext &context, DataChunk &
 
 	CreateFilterLocalSinkState &local_state = input.local_state.Cast<CreateFilterLocalSinkState>();
 	if (profiling_stats) {
-		ScopedTimer timer(profiling_stats->sink_time_us);
+		ScopedCpuTimer timer(profiling_stats->sink_time_us);
 		profiling_stats->rows_materialized.fetch_add(chunk.size(), std::memory_order_relaxed);
 		local_state.local_data->Append(chunk);
 	} else {
 		local_state.local_data->Append(chunk);
 	}
 
-	// insert into bloom filters
-	for (size_t i = 0; i < filter_operation->build_columns.size(); i++) {
-		const auto &col = filter_operation->build_columns[i];
-		auto it = bloom_filter_map.find(col);
-		if (it != bloom_filter_map.end() && it->second) {
-			it->second->Insert(chunk, {bound_column_indices[i]});
-		}
-	}
-
 	// compute min-max using typed pointer access
 	if (is_forward_pass && !local_state.local_min_max.empty() && chunk.size() > 0) {
 		for (idx_t i = 0; i < bound_column_indices.size() && i < local_state.local_min_max.size(); i++) {
@@ -455,6 +443,102 @@ static void PushDynamicFilters(const PhysicalCreateFilter &op, const CreateFilte
 	}
 }
 
+// parallelize the hashing/insert pass only when the build side is large enough to amortize
+// task-scheduling overhead; below this an inline single-threaded build is faster.
+static constexpr idx_t PARALLEL_BUILD_THRESHOLD = 102400; // ~100k rows (50 vectors)
+
+void PhysicalCreateFilter::PostBuildFinalize(const CreateFilterGlobalSinkState &gsink, ClientContext &context) const {
+	const idx_t actual_rows = gsink.total_data->Count();
+
+	for (auto &entry : bloom_filter_map) {
+		if (entry.second) {
+			entry.second->finalized_ = true;
+		}
+	}
+
+	// if this forward CREATE_FILTER produced an empty BF, signal sibling CREATE_FILTERs targeting
+	// the same probe that the probe side will be empty (relaxed store, lock-free).
+	if (gsink.probe_empty_flag && actual_rows == 0) {
+		gsink.probe_empty_flag->store(true, std::memory_order_relaxed);
+	}
+
+	PushDynamicFilters(*this, gsink, context);
+}
+
+namespace {
+
+// hashes + inserts a disjoint range of total_data chunks into the (already-allocated) BFs.
+// native BF inserts use atomic fetch_or, so concurrent inserts into the same filter are safe.
+class CreateFilterBuildTask : public ExecutorTask {
+public:
+	CreateFilterBuildTask(Pipeline &pipeline, shared_ptr<Event> event, const PhysicalCreateFilter &op,
+	                      CreateFilterGlobalSinkState &gsink, idx_t chunk_from, idx_t chunk_to)
+	    : ExecutorTask(pipeline.GetClientContext(), std::move(event), op), op(op), gsink(gsink), chunk_from(chunk_from),
+	      chunk_to(chunk_to) {
+	}
+
+	TaskExecutionResult ExecuteTask(TaskExecutionMode mode) override {
+		DataChunk chunk;
+		gsink.total_data->InitializeScanChunk(chunk);
+		for (idx_t cid = chunk_from; cid < chunk_to; cid++) {
+			gsink.total_data->FetchChunk(cid, chunk);
+			for (size_t i = 0; i < op.filter_operation->build_columns.size(); i++) {
+				auto it = op.bloom_filter_map.find(op.filter_operation->build_columns[i]);
+				if (it != op.bloom_filter_map.end() && it->second) {
+					it->second->Insert(chunk, {op.bound_column_indices[i]});
+				}
+			}
+		}
+		event->FinishTask();
+		return TaskExecutionResult::TASK_FINISHED;
+	}
+
+	string TaskType() const override {
+		return "CreateFilterBuildTask";
+	}
+
+private:
+	const PhysicalCreateFilter &op;
+	CreateFilterGlobalSinkState &gsink;
+	idx_t chunk_from;
+	idx_t chunk_to;
+};
+
+class CreateFilterBuildEvent : public BasePipelineEvent {
+public:
+	CreateFilterBuildEvent(Pipeline &pipeline, const PhysicalCreateFilter &op, CreateFilterGlobalSinkState &gsink)
+	    : BasePipelineEvent(pipeline), op(op), gsink(gsink) {
+	}
+
+	void Schedule() override {
+		auto &client = pipeline->GetClientContext();
+		const idx_t num_threads = NumericCast<idx_t>(TaskScheduler::GetScheduler(client).NumberOfThreads());
+		const idx_t chunk_count = gsink.total_data->ChunkCount();
+		const idx_t num_tasks = MaxValue<idx_t>(MinValue<idx_t>(num_threads, chunk_count), 1);
+		const idx_t chunks_per_task = (chunk_count + num_tasks - 1) / num_tasks;
+
+		vector<shared_ptr<Task>> tasks;
+		for (idx_t cid = 0; cid < chunk_count;) {
+			const idx_t from = cid;
+			const idx_t to = MinValue<idx_t>(from + chunks_per_task, chunk_count);
+			tasks.push_back(make_uniq<CreateFilterBuildTask>(*pipeline, shared_from_this(), op, gsink, from, to));
+			cid = to;
+		}
+		SetTasks(std::move(tasks));
+	}
+
+	void FinishEvent() override {
+		// runs after all build tasks complete, before dependent (probe) pipelines run
+		op.PostBuildFinalize(gsink, pipeline->GetClientContext());
+	}
+
+private:
+	const PhysicalCreateFilter &op;
+	CreateFilterGlobalSinkState &gsink;
+};
+
+} // namespace
+
 SinkFinalizeType PhysicalCreateFilter::Finalize(Pipeline &pipeline, Event &event, ClientContext &context,
                                                 OperatorSinkFinalizeInput &input) const {
 	// lazy init profiling if Sink was never called (e.g., empty input)
@@ -486,48 +570,36 @@ SinkFinalizeType PhysicalCreateFilter::Finalize(Pipeline &pipeline, Event &event
 	D_PRINTF("[FINALIZE] CREATE_FILTER (build=%s): total_data contains %llu rows, %zu bloom filters",
 	         build_table.c_str(), (unsigned long long)gsink.total_data->Count(), bloom_filter_map.size());
 
-	// 2. resize any undersized BFs and rehash from materialized data.
-	// rule: resize iff allocated_bits / actual_rows < 8  (i.e., <8 bits/key -> FPR > ~2%).
-	// shrink-on-overestimate is intentionally skipped.
-	// TODO - evaluate memory savings and performance tradeoff with shrink-on-overestimate
+	// 2. build bloom filters from the materialized data, sized to the actual row count
 	const idx_t actual_rows = gsink.total_data->Count();
+
+	// large build side: allocate up front (sized to actual rows), then hash/insert in parallel via
+	// scheduled tasks; post-build steps run from the event's completion.
+	if (actual_rows >= PARALLEL_BUILD_THRESHOLD && gsink.total_data->ChunkCount() > 1) {
+		for (size_t i = 0; i < filter_operation->build_columns.size(); i++) {
+			auto it = bloom_filter_map.find(filter_operation->build_columns[i]);
+			if (it != bloom_filter_map.end() && it->second) {
+				it->second->Initialize(context, actual_rows);
+			}
+		}
+		fin_timer.reset(); // insert time is spent on the scheduled tasks, not this thread
+		event.InsertEvent(make_shared_ptr<CreateFilterBuildEvent>(pipeline, *this, gsink));
+		return SinkFinalizeType::READY;
+	}
+
+	// small (or empty) build side: build inline on this thread
 	if (actual_rows > 0) {
 		for (size_t i = 0; i < filter_operation->build_columns.size(); i++) {
 			const auto &col = filter_operation->build_columns[i];
 			auto it = bloom_filter_map.find(col);
 			if (it == bloom_filter_map.end() || !it->second) {
 				continue;
 			}
-			auto &bf = *it->second;
-			const idx_t min_bits = std::max<idx_t>(512, bf.SizedForRows() * 12);
-			const idx_t allocated_bits = NextPowerOfTwo(min_bits);
-			if (actual_rows * 8 > allocated_bits) {
-				D_PRINTF("[RESIZE] CREATE_FILTER (build=%s) col=(%llu.%llu) sized_for=%llu actual=%llu "
-				         "allocated_bits=%llu -> rehashing",
-				         build_table.c_str(), (unsigned long long)col.table_index, (unsigned long long)col.column_index,
-				         (unsigned long long)bf.SizedForRows(), (unsigned long long)actual_rows,
-				         (unsigned long long)allocated_bits);
-				bf.ReinitializeAndRehash(context, actual_rows, *gsink.total_data, {bound_column_indices[i]});
-			}
+			it->second->ReinitializeAndRehash(context, actual_rows, *gsink.total_data, {bound_column_indices[i]});
 		}
 	}
 
-	// 3. mark bloom filters as finalized
-	for (auto &entry : bloom_filter_map) {
-		if (entry.second) {
-			entry.second->finalized_ = true;
-		}
-	}
-
-	// 4. if this forward CREATE_FILTER produced an empty BF, signal sibling CREATE_FILTERs
-	// targeting the same probe that the probe side will be empty (relaxed store, lock-free).
-	if (gsink.probe_empty_flag && actual_rows == 0) {
-		gsink.probe_empty_flag->store(true, std::memory_order_relaxed);
-	}
-
-	// 5. push dynamic filters to table scans
-	PushDynamicFilters(*this, gsink, context);
-
+	PostBuildFinalize(gsink, context);
 	return SinkFinalizeType::READY;
 }
 

src/operators/physical_create_filter.hpp

@@ -11,6 +11,7 @@
 namespace duckdb {
 
 struct CreateFilterStats;
+class CreateFilterGlobalSinkState;
 
 class PhysicalCreateFilter : public PhysicalOperator {
 public:
@@ -37,6 +38,10 @@ class PhysicalCreateFilter : public PhysicalOperator {
 	SinkFinalizeType Finalize(Pipeline &pipeline, Event &event, ClientContext &context,
 	                          OperatorSinkFinalizeInput &input) const override;
 
+	// mark filters finalized, signal empty probe, push dynamic filters; runs inline for small
+	// builds or from the parallel build event's completion for large builds.
+	void PostBuildFinalize(const CreateFilterGlobalSinkState &gsink, ClientContext &context) const;
+
 	bool IsSink() const override {
 		return true;
 	}

src/optimizer/robust_optimizer.cpp

@@ -1199,6 +1199,7 @@ RobustOptimizerContextState::BuildStackedBFOperators(unique_ptr<LogicalOperator>
 				// multiple consecutive CREATEs for same table - merge them
 				FilterOperation merged_op = consecutive_creates[0];
 				merged_op.build_columns.clear();
+				merged_op.probe_columns.clear();
 
 				// collect all build columns
 				for (const auto &op : consecutive_creates) {

src/robust_profiling.hpp

@@ -5,6 +5,7 @@
 #include "duckdb/common/printer.hpp"
 #include "duckdb/common/string_util.hpp"
 #include <chrono>
+#include <ctime>
 #include <atomic>
 #include <mutex>
 #include <vector>
@@ -49,6 +50,24 @@ struct ScopedTimer {
 	}
 };
 
+// RAII timer that adds elapsed *thread CPU* microseconds to an atomic counter.
+// uses CLOCK_THREAD_CPUTIME_ID so the measurement reflects work actually executed and
+// is not inflated by thread descheduling/contention (which wall-clock would capture).
+struct ScopedCpuTimer {
+	std::atomic<int64_t> &target;
+	timespec start;
+
+	explicit ScopedCpuTimer(std::atomic<int64_t> &target) : target(target) {
+		clock_gettime(CLOCK_THREAD_CPUTIME_ID, &start);
+	}
+	~ScopedCpuTimer() {
+		timespec end;
+		clock_gettime(CLOCK_THREAD_CPUTIME_ID, &end);
+		int64_t us = (end.tv_sec - start.tv_sec) * 1000000LL + (end.tv_nsec - start.tv_nsec) / 1000;
+		target.fetch_add(us, std::memory_order_relaxed);
+	}
+};
+
 class RobustProfilingState : public ClientContextState {
 public:
 	explicit RobustProfilingState(bool enabled) : enabled(enabled) {