diff --git a/extension_config.cmake b/extension_config.cmake index ea1b5e1..0140dc8 100644 --- a/extension_config.cmake +++ b/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) diff --git a/scripts/bench_tpch.sh b/scripts/bench_tpch.sh index c606b6d..f4f5532 100755 --- a/scripts/bench_tpch.sh +++ b/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 diff --git a/scripts/setup_tpch_data.sh b/scripts/setup_tpch_data.sh new file mode 100755 index 0000000..0f985b4 --- /dev/null +++ b/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" diff --git a/src/operators/physical_create_filter.cpp b/src/operators/physical_create_filter.cpp index 458fd91..2fb40b5 100644 --- a/src/operators/physical_create_filter.cpp +++ b/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(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(); 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, 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(TaskScheduler::GetScheduler(client).NumberOfThreads()); + const idx_t chunk_count = gsink.total_data->ChunkCount(); + const idx_t num_tasks = MaxValue(MinValue(num_threads, chunk_count), 1); + const idx_t chunks_per_task = (chunk_count + num_tasks - 1) / num_tasks; + + vector> tasks; + for (idx_t cid = 0; cid < chunk_count;) { + const idx_t from = cid; + const idx_t to = MinValue(from + chunks_per_task, chunk_count); + tasks.push_back(make_uniq(*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,11 +570,24 @@ 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(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]; @@ -498,36 +595,11 @@ SinkFinalizeType PhysicalCreateFilter::Finalize(Pipeline &pipeline, Event &event if (it == bloom_filter_map.end() || !it->second) { continue; } - auto &bf = *it->second; - const idx_t min_bits = std::max(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; } diff --git a/src/operators/physical_create_filter.hpp b/src/operators/physical_create_filter.hpp index 4b0d6ca..3d306c1 100644 --- a/src/operators/physical_create_filter.hpp +++ b/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; } diff --git a/src/optimizer/robust_optimizer.cpp b/src/optimizer/robust_optimizer.cpp index a80da57..567e5ef 100644 --- a/src/optimizer/robust_optimizer.cpp +++ b/src/optimizer/robust_optimizer.cpp @@ -1199,6 +1199,7 @@ RobustOptimizerContextState::BuildStackedBFOperators(unique_ptr // 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) { diff --git a/src/robust_profiling.hpp b/src/robust_profiling.hpp index 206ac04..1909c5b 100644 --- a/src/robust_profiling.hpp +++ b/src/robust_profiling.hpp @@ -5,6 +5,7 @@ #include "duckdb/common/printer.hpp" #include "duckdb/common/string_util.hpp" #include +#include #include #include #include @@ -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 ⌖ + timespec start; + + explicit ScopedCpuTimer(std::atomic &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) {