From 380a4b5e9bbb5d8557a1636bcb985af6de8d7143 Mon Sep 17 00:00:00 2001 From: ila Date: Tue, 1 Sep 2026 11:56:31 +0200 Subject: [PATCH 1/6] Fuse filterless floating and count partials --- src/aggregates/filterless_aggregate.cpp | 208 +++++++++++++++++++++++- src/compiler/privacy_mechanisms.cpp | 131 ++++++++++----- test/sql/dp_filterless.test | 56 +++++++ 3 files changed, 353 insertions(+), 42 deletions(-) diff --git a/src/aggregates/filterless_aggregate.cpp b/src/aggregates/filterless_aggregate.cpp index d44e054..bf96a26 100644 --- a/src/aggregates/filterless_aggregate.cpp +++ b/src/aggregates/filterless_aggregate.cpp @@ -937,9 +937,8 @@ struct FilterlessApproxSumOperation { return Hugeint::Convert(ClipApproximateMagnitude64(AsScaledMagnitude(value))); } - template - static void Operation(STATE &state, const INPUT_TYPE &input, AggregateUnaryInput &) { - double value = static_cast(input); + template + static void AddValue(STATE &state, double value) { if (!std::isfinite(value)) { throw InvalidInputException("filterless: per-PU SUM contribution must be finite"); } @@ -952,6 +951,11 @@ struct FilterlessApproxSumOperation { } } + template + static void Operation(STATE &state, const INPUT_TYPE &input, AggregateUnaryInput &) { + AddValue(state, static_cast(input)); + } + template static void ConstantOperation(STATE &state, const INPUT_TYPE &input, AggregateUnaryInput &, idx_t count) { double value = static_cast(input); @@ -974,14 +978,19 @@ struct FilterlessApproxSumOperation { target.negative = Hugeint::Add(target.negative, source.negative); } + template + static double FinalizeValue(const STATE &state) { + auto scaled = Hugeint::Subtract(state.positive, state.negative); + return Hugeint::Cast(scaled) / CLIP_DOUBLE_SCALE; + } + template static void Finalize(STATE &state, RESULT_TYPE &target, AggregateFinalizeData &finalize_data) { if (!state.isset) { finalize_data.ReturnNull(); return; } - auto scaled = Hugeint::Subtract(state.positive, state.negative); - target = Hugeint::Cast(scaled) / CLIP_DOUBLE_SCALE; + target = FinalizeValue(state); } static bool IgnoreNull() { @@ -989,6 +998,172 @@ struct FilterlessApproxSumOperation { } }; +struct FilterlessApproxSumPairState { + FilterlessApproxSumState answer; + FilterlessApproxSumState histogram; +}; + +struct FilterlessCountPairState { + uint64_t answer; + uint64_t histogram; +}; + +static LogicalType FilterlessLowerPairType(const LogicalType &value_type) { + child_list_t children; + children.emplace_back("answer", value_type); + children.emplace_back("histogram", value_type); + return LogicalType::STRUCT(std::move(children)); +} + +static idx_t FilterlessApproxSumPairStateSize(const AggregateFunction &) { + return sizeof(FilterlessApproxSumPairState); +} + +static idx_t FilterlessCountPairStateSize(const AggregateFunction &) { + return sizeof(FilterlessCountPairState); +} + +static void FilterlessApproxSumPairInitialize(const AggregateFunction &, data_ptr_t state_p) { + auto &state = *reinterpret_cast(state_p); + FilterlessApproxSumOperation::Initialize(state.answer); + FilterlessApproxSumOperation::Initialize(state.histogram); +} + +static void FilterlessCountPairInitialize(const AggregateFunction &, data_ptr_t state_p) { + memset(state_p, 0, sizeof(FilterlessCountPairState)); +} + +template +static void FilterlessLowerPairUpdateRows(Vector inputs[], idx_t count, STATE_GETTER get_state, UPDATE update) { + UnifiedVectorFormat value_data; + UnifiedVectorFormat active_data; + UnifiedVectorFormat sampled_data; + inputs[0].ToUnifiedFormat(count, value_data); + inputs[1].ToUnifiedFormat(count, active_data); + inputs[2].ToUnifiedFormat(count, sampled_data); + auto values = UnifiedVectorFormat::GetData(value_data); + auto active = UnifiedVectorFormat::GetData(active_data); + auto sampled = UnifiedVectorFormat::GetData(sampled_data); + for (idx_t row = 0; row < count; row++) { + auto value_index = value_data.sel->get_index(row); + if (!value_data.validity.RowIsValid(value_index)) { + continue; + } + auto active_index = active_data.sel->get_index(row); + auto sampled_index = sampled_data.sel->get_index(row); + bool is_active = active_data.validity.RowIsValid(active_index) && active[active_index]; + bool is_sampled = sampled_data.validity.RowIsValid(sampled_index) && sampled[sampled_index]; + if (!is_active && !is_sampled) { + continue; + } + update(*get_state(row), values[value_index], is_active, is_sampled); + } +} + +static void UpdateFilterlessApproxSumPair(FilterlessApproxSumPairState &state, double value, bool active, + bool sampled) { + if (active) { + FilterlessApproxSumOperation::AddValue(state.answer, value); + } + if (sampled) { + FilterlessApproxSumOperation::AddValue(state.histogram, value); + } +} + +static void UpdateFilterlessCountPair(FilterlessCountPairState &state, bool count_value, bool active, bool sampled) { + if (!count_value) { + return; + } + state.answer += static_cast(active); + state.histogram += static_cast(sampled); +} + +static void FilterlessApproxSumPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, + idx_t count) { + auto state = reinterpret_cast(state_p); + FilterlessLowerPairUpdateRows( + inputs, count, [state](idx_t) { return state; }, UpdateFilterlessApproxSumPair); +} + +static void FilterlessCountPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, idx_t count) { + auto state = reinterpret_cast(state_p); + FilterlessLowerPairUpdateRows( + inputs, count, [state](idx_t) { return state; }, UpdateFilterlessCountPair); +} + +template +static void FilterlessLowerPairScatterUpdate(Vector inputs[], Vector &states, idx_t count, UPDATE update) { + UnifiedVectorFormat state_data; + states.ToUnifiedFormat(count, state_data); + auto state_ptrs = UnifiedVectorFormat::GetData(state_data); + FilterlessLowerPairUpdateRows( + inputs, count, [&](idx_t row) { return state_ptrs[state_data.sel->get_index(row)]; }, update); +} + +static void FilterlessApproxSumPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, + idx_t count) { + FilterlessLowerPairScatterUpdate(inputs, states, count, + UpdateFilterlessApproxSumPair); +} + +static void FilterlessCountPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, + idx_t count) { + FilterlessLowerPairScatterUpdate(inputs, states, count, UpdateFilterlessCountPair); +} + +static void FilterlessApproxSumPairCombine(Vector &source, Vector &target, AggregateInputData &input, idx_t count) { + auto sources = FlatVector::GetData(source); + auto targets = FlatVector::GetData(target); + for (idx_t i = 0; i < count; i++) { + FilterlessApproxSumOperation::Combine( + sources[i]->answer, targets[i]->answer, input); + FilterlessApproxSumOperation::Combine( + sources[i]->histogram, targets[i]->histogram, input); + } +} + +static void FilterlessCountPairCombine(Vector &source, Vector &target, AggregateInputData &, idx_t count) { + auto sources = FlatVector::GetData(source); + auto targets = FlatVector::GetData(target); + for (idx_t i = 0; i < count; i++) { + targets[i]->answer += sources[i]->answer; + targets[i]->histogram += sources[i]->histogram; + } +} + +static void FilterlessApproxSumPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, + idx_t offset) { + auto state_ptrs = FlatVector::GetData(states); + auto &children = StructVector::GetEntries(result); + auto answers = FlatVector::GetData(*children[0]); + auto histograms = FlatVector::GetData(*children[1]); + for (idx_t i = 0; i < count; i++) { + auto row = offset + i; + if (state_ptrs[i]->answer.isset) { + answers[row] = FilterlessApproxSumOperation::FinalizeValue(state_ptrs[i]->answer); + } else { + FlatVector::Validity(*children[0]).SetInvalid(row); + } + if (state_ptrs[i]->histogram.isset) { + histograms[row] = FilterlessApproxSumOperation::FinalizeValue(state_ptrs[i]->histogram); + } else { + FlatVector::Validity(*children[1]).SetInvalid(row); + } + } +} + +static void FilterlessCountPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, + idx_t offset) { + auto state_ptrs = FlatVector::GetData(states); + auto &children = StructVector::GetEntries(result); + auto answers = FlatVector::GetData(*children[0]); + auto histograms = FlatVector::GetData(*children[1]); + for (idx_t i = 0; i < count; i++) { + answers[offset + i] = static_cast(state_ptrs[i]->answer); + histograms[offset + i] = static_cast(state_ptrs[i]->histogram); + } +} + static LogicalType FilterlessDebugType() { child_list_t children; children.emplace_back("lower_bound", LogicalType::DOUBLE); @@ -1126,6 +1301,29 @@ void RegisterFilterlessAggregateFunctions(ExtensionLoader &loader) { approx_sum_info.descriptions.push_back(std::move(approx_sum_description)); loader.RegisterFunction(std::move(approx_sum_info)); + AggregateFunction approx_sum_pair( + "priv_filterless_approx_sum_pair", {LogicalType::DOUBLE, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, + FilterlessLowerPairType(LogicalType::DOUBLE), FilterlessApproxSumPairStateSize, + FilterlessApproxSumPairInitialize, FilterlessApproxSumPairScatterUpdate, FilterlessApproxSumPairCombine, + FilterlessApproxSumPairFinalize, FunctionNullHandling::SPECIAL_HANDLING, FilterlessApproxSumPairUpdate); + CreateAggregateFunctionInfo approx_sum_pair_info(approx_sum_pair); + FunctionDescription approx_sum_pair_description; + approx_sum_pair_description.description = + "[INTERNAL] Fused filtered-answer and sampled-histogram approximate SUM partials."; + approx_sum_pair_info.descriptions.push_back(std::move(approx_sum_pair_description)); + loader.RegisterFunction(std::move(approx_sum_pair_info)); + + AggregateFunction count_pair( + "priv_filterless_count_pair", {LogicalType::BOOLEAN, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, + FilterlessLowerPairType(LogicalType::BIGINT), FilterlessCountPairStateSize, FilterlessCountPairInitialize, + FilterlessCountPairScatterUpdate, FilterlessCountPairCombine, FilterlessCountPairFinalize, + FunctionNullHandling::SPECIAL_HANDLING, FilterlessCountPairUpdate); + CreateAggregateFunctionInfo count_pair_info(count_pair); + FunctionDescription count_pair_description; + count_pair_description.description = "[INTERNAL] Fused filtered-answer and sampled-histogram COUNT partials."; + count_pair_info.descriptions.push_back(std::move(count_pair_description)); + loader.RegisterFunction(std::move(count_pair_info)); + AggregateFunctionSet sum_set("filterless_sum"); AddSumCountOverloads(sum_set, "filterless_sum", FilterlessFinalize, LogicalType::DOUBLE, BindFilterlessSum); diff --git a/src/compiler/privacy_mechanisms.cpp b/src/compiler/privacy_mechanisms.cpp index 280b777..b229969 100644 --- a/src/compiler/privacy_mechanisms.cpp +++ b/src/compiler/privacy_mechanisms.cpp @@ -3166,14 +3166,21 @@ static unique_ptr BuildFilterlessEncodedPuInput(OptimizerExtensionIn return make_uniq(LogicalType::UBIGINT, propagated); } +struct FilterlessLowerComponentInput { + idx_t answer_column; + idx_t histogram_column; + bool paired; +}; + struct FilterlessPreAggregationInput { idx_t group_table_index; idx_t aggregate_table_index; idx_t group_count; idx_t component_column_offset; - idx_t component_count; + idx_t active_count_column; vector group_types; vector aggregate_types; + vector components; unique_ptr GroupRef(idx_t index) const { return make_uniq(group_types[index], ColumnBinding(group_table_index, index)); @@ -3188,48 +3195,64 @@ struct FilterlessPreAggregationInput { aggregate_types[index], ColumnBinding(aggregate_table_index, component_column_offset + index)); } - unique_ptr AnswerRef(idx_t index) const { - return AggregateRef(index); + unique_ptr ComponentRef(OptimizerExtensionInput &input, idx_t index, bool answer) const { + auto &component = components[index]; + auto column = answer ? component.answer_column : component.histogram_column; + auto value = AggregateRef(column); + if (!component.paired) { + return value; + } + vector> children; + children.push_back(std::move(value)); + children.push_back(make_uniq(Value(answer ? "answer" : "histogram"))); + return BindScalarLocal(input, "struct_extract", std::move(children)); } - unique_ptr HistogramRef(idx_t index) const { - return AggregateRef(component_count + index); + unique_ptr AnswerRef(OptimizerExtensionInput &input, idx_t index) const { + return ComponentRef(input, index, true); + } + + unique_ptr HistogramRef(OptimizerExtensionInput &input, idx_t index) const { + return ComponentRef(input, index, false); } unique_ptr ActiveCountRef() const { - return AggregateRef(2 * component_count); + return AggregateRef(active_count_column); } }; -static FilterlessPreAggregationInput BuildFilterlessPreAggregationInput(const PuPreAggregationInfo &pre, - idx_t component_count) { +static FilterlessPreAggregationInput +BuildFilterlessPreAggregationInput(const PuPreAggregationInfo &pre, vector components, + idx_t active_count_column) { FilterlessPreAggregationInput result; result.group_table_index = pre.lower_agg->group_index; result.aggregate_table_index = pre.lower_agg_index; result.group_count = pre.num_original_groups; result.component_column_offset = 0; - result.component_count = component_count; + result.active_count_column = active_count_column; + result.components = std::move(components); for (idx_t i = 0; i < pre.num_original_groups; i++) { result.group_types.push_back(pre.lower_agg->types[i]); } - for (idx_t i = 0; i < 2 * component_count + 1; i++) { + for (idx_t i = 0; i < pre.lower_agg->expressions.size(); i++) { result.aggregate_types.push_back(pre.lower_agg->types[pre.num_original_groups + 1 + i]); } return result; } static FilterlessPreAggregationInput ApplyFilterlessMaxGroups(OptimizerExtensionInput &input, LogicalAggregate *agg, - const PuPreAggregationInfo &pre, idx_t component_count, - int64_t max_groups) { + const PuPreAggregationInfo &pre, + vector components, + idx_t active_count_column, int64_t max_groups) { if (pre.num_original_groups == 0) { - return BuildFilterlessPreAggregationInput(pre, component_count); + return BuildFilterlessPreAggregationInput(pre, std::move(components), active_count_column); } // Flatten the lower aggregate's separate group/aggregate bindings into one projection. The // lower aggregate already emits exactly one row per logical (PU, SQL group), so ROW_NUMBER // caps the group set without needing duplicate-aware DENSE_RANK. DuckDB can optimize this // single hashed ordering into a bounded per-PU top-k aggregate. - idx_t lower_aggregate_count = 2 * component_count + 1; + idx_t lower_aggregate_count = pre.lower_agg->expressions.size(); idx_t projection_index = input.optimizer.binder.GenerateTableIndex(); vector> expressions; expressions.reserve(pre.num_original_groups + 1 + lower_aggregate_count); @@ -3272,7 +3295,8 @@ static FilterlessPreAggregationInput ApplyFilterlessMaxGroups(OptimizerExtension result.aggregate_table_index = projection_index; result.group_count = pre.num_original_groups; result.component_column_offset = pre.num_original_groups + 1; - result.component_count = component_count; + result.active_count_column = active_count_column; + result.components = std::move(components); for (idx_t i = 0; i < pre.num_original_groups; i++) { result.group_types.push_back(pre.lower_agg->types[i]); } @@ -3341,19 +3365,41 @@ static unique_ptr BuildFilterlessLogicalPu(OptimizerExtensionInput & make_uniq(Value::UBIGINT(~uint64_t(1)))); } -static unique_ptr BuildFilterlessLowerAggregate(OptimizerExtensionInput &input, - const BoundAggregateExpression &aggregate, bool is_count) { +static unique_ptr BuildFilterlessExactLowerSumAggregate(OptimizerExtensionInput &input, + const BoundAggregateExpression &aggregate) { + auto input_type = aggregate.children[0]->return_type.InternalType(); + D_ASSERT(input_type != PhysicalType::FLOAT && input_type != PhysicalType::DOUBLE); + return BindPlainAggregate(input, "sum", aggregate.children[0]->Copy()); +} + +static bool CanFuseFilterlessLowerAggregate(const BoundAggregateExpression &aggregate, bool is_count) { if (is_count) { - if (aggregate.function.name == "count" && !aggregate.children.empty()) { - return BindPlainAggregate(input, "count", aggregate.children[0]->Copy()); - } - return BindPlainAggregate(input, "count_star", nullptr); + return true; } auto input_type = aggregate.children[0]->return_type.InternalType(); - if (input_type == PhysicalType::FLOAT || input_type == PhysicalType::DOUBLE) { - return BindPlainAggregate(input, "filterless_approx_sum", aggregate.children[0]->Copy()); + return input_type == PhysicalType::FLOAT || input_type == PhysicalType::DOUBLE; +} + +static unique_ptr BuildFilterlessLowerPair(OptimizerExtensionInput &input, + const BoundAggregateExpression &aggregate, bool is_count, + const Expression &encoded_pu, int sample_bits) { + vector> children; + if (is_count) { + if (aggregate.function.name == "count" && !aggregate.children.empty()) { + auto countable = + make_uniq(ExpressionType::OPERATOR_IS_NOT_NULL, LogicalType::BOOLEAN); + countable->children.push_back(aggregate.children[0]->Copy()); + children.push_back(std::move(countable)); + } else { + children.push_back(make_uniq(Value::BOOLEAN(true))); + } + } else { + children.push_back(aggregate.children[0]->Copy()); } - return BindPlainAggregate(input, "sum", aggregate.children[0]->Copy()); + children.push_back(BuildFilterlessIsActive(input, encoded_pu)); + children.push_back(BuildFilterlessSamplePredicate(input, encoded_pu, sample_bits)); + return BindAggregateLocal(input, is_count ? "priv_filterless_count_pair" : "priv_filterless_approx_sum_pair", + std::move(children)); } void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, OptimizerExtensionInput &input, @@ -3423,37 +3469,48 @@ void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, Optimizer vector> lower_expressions; count_components.reserve(component_count); lower_expressions.reserve(2 * component_count + 1); - vector> histogram_expressions; - histogram_expressions.reserve(component_count); + vector lower_components; + lower_components.reserve(component_count); for (idx_t i = 0; i < component_count; i++) { auto &aggregate = agg->expressions[i]->Cast(); bool is_count = IsCountAggregate(aggregate); count_components.push_back(is_count); - auto answer_partial = BuildFilterlessLowerAggregate(input, aggregate, is_count); + if (CanFuseFilterlessLowerAggregate(aggregate, is_count)) { + auto column = lower_expressions.size(); + lower_expressions.push_back( + BuildFilterlessLowerPair(input, aggregate, is_count, *encoded_pu, filterless_settings.sample_bits)); + lower_components.push_back({column, column, true}); + continue; + } + + D_ASSERT(!is_count); + auto answer_column = lower_expressions.size(); + auto answer_partial = BuildFilterlessExactLowerSumAggregate(input, aggregate); answer_partial->Cast().filter = BuildFilterlessIsActive(input, *encoded_pu); lower_expressions.push_back(std::move(answer_partial)); - auto histogram_partial = BuildFilterlessLowerAggregate(input, aggregate, is_count); + auto histogram_column = lower_expressions.size(); + auto histogram_partial = BuildFilterlessExactLowerSumAggregate(input, aggregate); histogram_partial->Cast().filter = BuildFilterlessSamplePredicate(input, *encoded_pu, filterless_settings.sample_bits); - histogram_expressions.push_back(std::move(histogram_partial)); - } - for (auto &histogram_expression : histogram_expressions) { - lower_expressions.push_back(std::move(histogram_expression)); + lower_expressions.push_back(std::move(histogram_partial)); + lower_components.push_back({answer_column, histogram_column, false}); } + idx_t active_count_column = lower_expressions.size(); auto active_count = BindPlainAggregate(input, "count_star", nullptr); active_count->Cast().filter = BuildFilterlessIsActive(input, *encoded_pu); lower_expressions.push_back(std::move(active_count)); auto logical_pu = BuildFilterlessLogicalPu(input, std::move(encoded_pu)); auto pre = InsertPuPreAggregation(input, agg, std::move(lower_expressions), std::move(logical_pu)); - PRIVACY_DEBUG_PRINT("[dp_filterless] pre-aggregated separate filtered-answer and fixed-sample histogram " - "contributions; floating SUM uses the scalar AS magnitude accumulator"); - auto pre_input = ApplyFilterlessMaxGroups(input, agg, pre, component_count, max_groups); + PRIVACY_DEBUG_PRINT("[dp_filterless] pre-aggregated paired filtered-answer and fixed-sample histogram " + "contributions where supported; floating SUM uses the scalar AS magnitude accumulator"); + auto pre_input = + ApplyFilterlessMaxGroups(input, agg, pre, std::move(lower_components), active_count_column, max_groups); double visible_cell_epsilon = epsilon / budget_units; for (idx_t i = 0; i < component_count; i++) { double component_epsilon = avg_components.count(i) ? visible_cell_epsilon / 2.0 : visible_cell_epsilon; - unique_ptr answer_partial = pre_input.AnswerRef(i); - unique_ptr histogram_partial = pre_input.HistogramRef(i); + unique_ptr answer_partial = pre_input.AnswerRef(input, i); + unique_ptr histogram_partial = pre_input.HistogramRef(input, i); auto is_active = make_uniq(ExpressionType::COMPARE_GREATERTHAN, pre_input.ActiveCountRef(), make_uniq(Value::BIGINT(0))); diff --git a/test/sql/dp_filterless.test b/test/sql/dp_filterless.test index de06e1c..e584827 100644 --- a/test/sql/dp_filterless.test +++ b/test/sql/dp_filterless.test @@ -77,6 +77,36 @@ FROM ( ---- 25.000000 100.000000 4.000000 +# The internal lower aggregates compute filtered-answer and sampled-histogram +# partials together. NULL values retain SUM and COUNT's ordinary semantics. +query RR +SELECT x.answer, x.histogram +FROM ( + SELECT priv_filterless_approx_sum_pair(value, active, sampled) AS x + FROM (VALUES + (10.0::DOUBLE, true, true), + (20.0::DOUBLE, false, true), + (NULL::DOUBLE, true, true), + (30.0::DOUBLE, true, false) + ) input(value, active, sampled) +); +---- +40.000000 30.000000 + +query II +SELECT x.answer, x.histogram +FROM ( + SELECT priv_filterless_count_pair(countable, active, sampled) AS x + FROM (VALUES + (true, true, true), + (false, true, true), + (true, false, true), + (true, true, false) + ) input(countable, active, sampled) +); +---- +2 2 + # Sampling is deterministic from the high PU bits. At p=6, PU 0 is sampled and # receives weight 64; PUs with the high bit set are not sampled. Qualifying # non-sampled PUs enter the answer but not the histogram. @@ -501,6 +531,32 @@ WHERE NOT qualifies; ---- 2000.000000 2 1000.000000 +# The compiler uses one paired lower aggregate per floating SUM/COUNT +# component, instead of separate filtered and sampled aggregates. +statement ok +SET explain_output = 'physical_only'; + +query II +EXPLAIN SELECT SUM(amount), COUNT(*) +FROM filterless_sampled_events +WHERE qualifies; +---- +physical_plan :[\s\S]*priv_filterless_approx_sum_[\s\S]*pair[\s\S]*priv_filterless_count_pair[\s\S]* + +# COUNT(expr) fuses through an explicit non-NULL marker, while COUNT(*) still +# counts active rows regardless of the expression's NULL value. +statement ok +INSERT INTO filterless_sampled_events VALUES + (0, true, NULL), + (89, true, NULL); + +query RIIR +SELECT SUM(amount), COUNT(amount), COUNT(*), AVG(amount) +FROM filterless_sampled_events +WHERE qualifies; +---- +10000.000000 2 4 5000.000000 + statement ok SET dp_filterless_sample_bits = 0; From cb6d47676dfae5167b9666d87e7f1195678bbf72 Mon Sep 17 00:00:00 2001 From: ila Date: Tue, 1 Sep 2026 12:03:48 +0200 Subject: [PATCH 2/6] Format filterless fusion changes --- src/aggregates/filterless_aggregate.cpp | 12 ++++++------ src/compiler/privacy_mechanisms.cpp | 14 +++++++------- 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/src/aggregates/filterless_aggregate.cpp b/src/aggregates/filterless_aggregate.cpp index bf96a26..2f570bc 100644 --- a/src/aggregates/filterless_aggregate.cpp +++ b/src/aggregates/filterless_aggregate.cpp @@ -1061,7 +1061,7 @@ static void FilterlessLowerPairUpdateRows(Vector inputs[], idx_t count, STATE_GE } static void UpdateFilterlessApproxSumPair(FilterlessApproxSumPairState &state, double value, bool active, - bool sampled) { + bool sampled) { if (active) { FilterlessApproxSumOperation::AddValue(state.answer, value); } @@ -1079,7 +1079,7 @@ static void UpdateFilterlessCountPair(FilterlessCountPairState &state, bool coun } static void FilterlessApproxSumPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, - idx_t count) { + idx_t count) { auto state = reinterpret_cast(state_p); FilterlessLowerPairUpdateRows( inputs, count, [state](idx_t) { return state; }, UpdateFilterlessApproxSumPair); @@ -1101,13 +1101,13 @@ static void FilterlessLowerPairScatterUpdate(Vector inputs[], Vector &states, id } static void FilterlessApproxSumPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, - idx_t count) { + idx_t count) { FilterlessLowerPairScatterUpdate(inputs, states, count, UpdateFilterlessApproxSumPair); } static void FilterlessCountPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, - idx_t count) { + idx_t count) { FilterlessLowerPairScatterUpdate(inputs, states, count, UpdateFilterlessCountPair); } @@ -1132,7 +1132,7 @@ static void FilterlessCountPairCombine(Vector &source, Vector &target, Aggregate } static void FilterlessApproxSumPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, - idx_t offset) { + idx_t offset) { auto state_ptrs = FlatVector::GetData(states); auto &children = StructVector::GetEntries(result); auto answers = FlatVector::GetData(*children[0]); @@ -1153,7 +1153,7 @@ static void FilterlessApproxSumPairFinalize(Vector &states, AggregateInputData & } static void FilterlessCountPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, - idx_t offset) { + idx_t offset) { auto state_ptrs = FlatVector::GetData(states); auto &children = StructVector::GetEntries(result); auto answers = FlatVector::GetData(*children[0]); diff --git a/src/compiler/privacy_mechanisms.cpp b/src/compiler/privacy_mechanisms.cpp index b229969..ffcd312 100644 --- a/src/compiler/privacy_mechanisms.cpp +++ b/src/compiler/privacy_mechanisms.cpp @@ -3223,7 +3223,7 @@ struct FilterlessPreAggregationInput { static FilterlessPreAggregationInput BuildFilterlessPreAggregationInput(const PuPreAggregationInfo &pre, vector components, - idx_t active_count_column) { + idx_t active_count_column) { FilterlessPreAggregationInput result; result.group_table_index = pre.lower_agg->group_index; result.aggregate_table_index = pre.lower_agg_index; @@ -3241,9 +3241,9 @@ BuildFilterlessPreAggregationInput(const PuPreAggregationInfo &pre, vector components, - idx_t active_count_column, int64_t max_groups) { + const PuPreAggregationInfo &pre, + vector components, + idx_t active_count_column, int64_t max_groups) { if (pre.num_original_groups == 0) { return BuildFilterlessPreAggregationInput(pre, std::move(components), active_count_column); } @@ -3366,7 +3366,7 @@ static unique_ptr BuildFilterlessLogicalPu(OptimizerExtensionInput & } static unique_ptr BuildFilterlessExactLowerSumAggregate(OptimizerExtensionInput &input, - const BoundAggregateExpression &aggregate) { + const BoundAggregateExpression &aggregate) { auto input_type = aggregate.children[0]->return_type.InternalType(); D_ASSERT(input_type != PhysicalType::FLOAT && input_type != PhysicalType::DOUBLE); return BindPlainAggregate(input, "sum", aggregate.children[0]->Copy()); @@ -3381,8 +3381,8 @@ static bool CanFuseFilterlessLowerAggregate(const BoundAggregateExpression &aggr } static unique_ptr BuildFilterlessLowerPair(OptimizerExtensionInput &input, - const BoundAggregateExpression &aggregate, bool is_count, - const Expression &encoded_pu, int sample_bits) { + const BoundAggregateExpression &aggregate, bool is_count, + const Expression &encoded_pu, int sample_bits) { vector> children; if (is_count) { if (aggregate.function.name == "count" && !aggregate.children.empty()) { From 4fe9df7a1b827d71d61c8a8bf57ddbf469c542fe Mon Sep 17 00:00:00 2001 From: ila Date: Tue, 1 Sep 2026 12:25:36 +0200 Subject: [PATCH 3/6] Fuse exact filterless sum partials --- src/aggregates/filterless_aggregate.cpp | 171 ++++++++++++++++++++++++ src/compiler/privacy_mechanisms.cpp | 90 ++++--------- test/sql/dp_filterless.test | 55 +++++++- 3 files changed, 252 insertions(+), 64 deletions(-) diff --git a/src/aggregates/filterless_aggregate.cpp b/src/aggregates/filterless_aggregate.cpp index 2f570bc..13252fe 100644 --- a/src/aggregates/filterless_aggregate.cpp +++ b/src/aggregates/filterless_aggregate.cpp @@ -1008,6 +1008,26 @@ struct FilterlessCountPairState { uint64_t histogram; }; +struct FilterlessExactSumPartialState { + bool isset; + hugeint_t value; + + void Initialize() { + isset = false; + value = hugeint_t(0); + } + + void Combine(const FilterlessExactSumPartialState &other) { + isset = isset || other.isset; + value = Hugeint::Add(value, other.value); + } +}; + +struct FilterlessExactSumPairState { + FilterlessExactSumPartialState answer; + FilterlessExactSumPartialState histogram; +}; + static LogicalType FilterlessLowerPairType(const LogicalType &value_type) { child_list_t children; children.emplace_back("answer", value_type); @@ -1023,6 +1043,10 @@ static idx_t FilterlessCountPairStateSize(const AggregateFunction &) { return sizeof(FilterlessCountPairState); } +static idx_t FilterlessExactSumPairStateSize(const AggregateFunction &) { + return sizeof(FilterlessExactSumPairState); +} + static void FilterlessApproxSumPairInitialize(const AggregateFunction &, data_ptr_t state_p) { auto &state = *reinterpret_cast(state_p); FilterlessApproxSumOperation::Initialize(state.answer); @@ -1033,6 +1057,12 @@ static void FilterlessCountPairInitialize(const AggregateFunction &, data_ptr_t memset(state_p, 0, sizeof(FilterlessCountPairState)); } +static void FilterlessExactSumPairInitialize(const AggregateFunction &, data_ptr_t state_p) { + auto &state = *reinterpret_cast(state_p); + state.answer.Initialize(); + state.histogram.Initialize(); +} + template static void FilterlessLowerPairUpdateRows(Vector inputs[], idx_t count, STATE_GETTER get_state, UPDATE update) { UnifiedVectorFormat value_data; @@ -1078,6 +1108,43 @@ static void UpdateFilterlessCountPair(FilterlessCountPairState &state, bool coun state.histogram += static_cast(sampled); } +template +static void AddFilterlessExactInteger(hugeint_t &result, INPUT_TYPE value) { + // Match DuckDB's exact integer SUM accumulator: add the two's-complement low + // word and adjust the high word only on carry or borrow. + auto lower = static_cast(value); + result.lower += lower; + int overflow = result.lower < lower; + int positive = value >= 0; + if (!(overflow ^ positive)) { + result.upper += -1 + 2 * positive; + } +} + +template <> +void AddFilterlessExactInteger(hugeint_t &result, hugeint_t value) { + result = Hugeint::Add(result, value); +} + +template +static void AddFilterlessExactSumPairValue(FilterlessExactSumPairState &state, INPUT_TYPE value, bool active, + bool sampled) { + if (active) { + state.answer.isset = true; + AddFilterlessExactInteger(state.answer.value, value); + } + if (sampled) { + state.histogram.isset = true; + AddFilterlessExactInteger(state.histogram.value, value); + } +} + +template +static void UpdateFilterlessExactSumPair(FilterlessExactSumPairState &state, INPUT_TYPE value, bool active, + bool sampled) { + AddFilterlessExactSumPairValue(state, value, active, sampled); +} + static void FilterlessApproxSumPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, idx_t count) { auto state = reinterpret_cast(state_p); @@ -1091,6 +1158,14 @@ static void FilterlessCountPairUpdate(Vector inputs[], AggregateInputData &, idx inputs, count, [state](idx_t) { return state; }, UpdateFilterlessCountPair); } +template +static void FilterlessExactSumPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, + idx_t count) { + auto state = reinterpret_cast(state_p); + FilterlessLowerPairUpdateRows( + inputs, count, [state](idx_t) { return state; }, UpdateFilterlessExactSumPair); +} + template static void FilterlessLowerPairScatterUpdate(Vector inputs[], Vector &states, idx_t count, UPDATE update) { UnifiedVectorFormat state_data; @@ -1111,6 +1186,13 @@ static void FilterlessCountPairScatterUpdate(Vector inputs[], AggregateInputData FilterlessLowerPairScatterUpdate(inputs, states, count, UpdateFilterlessCountPair); } +template +static void FilterlessExactSumPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, + idx_t count) { + FilterlessLowerPairScatterUpdate(inputs, states, count, + UpdateFilterlessExactSumPair); +} + static void FilterlessApproxSumPairCombine(Vector &source, Vector &target, AggregateInputData &input, idx_t count) { auto sources = FlatVector::GetData(source); auto targets = FlatVector::GetData(target); @@ -1131,6 +1213,15 @@ static void FilterlessCountPairCombine(Vector &source, Vector &target, Aggregate } } +static void FilterlessExactSumPairCombine(Vector &source, Vector &target, AggregateInputData &, idx_t count) { + auto sources = FlatVector::GetData(source); + auto targets = FlatVector::GetData(target); + for (idx_t i = 0; i < count; i++) { + targets[i]->answer.Combine(sources[i]->answer); + targets[i]->histogram.Combine(sources[i]->histogram); + } +} + static void FilterlessApproxSumPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, idx_t offset) { auto state_ptrs = FlatVector::GetData(states); @@ -1164,6 +1255,27 @@ static void FilterlessCountPairFinalize(Vector &states, AggregateInputData &, Ve } } +static void FilterlessExactSumPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, + idx_t offset) { + auto state_ptrs = FlatVector::GetData(states); + auto &children = StructVector::GetEntries(result); + auto answers = FlatVector::GetData(*children[0]); + auto histograms = FlatVector::GetData(*children[1]); + for (idx_t i = 0; i < count; i++) { + auto row = offset + i; + if (state_ptrs[i]->answer.isset) { + answers[row] = state_ptrs[i]->answer.value; + } else { + FlatVector::Validity(*children[0]).SetInvalid(row); + } + if (state_ptrs[i]->histogram.isset) { + histograms[row] = state_ptrs[i]->histogram.value; + } else { + FlatVector::Validity(*children[1]).SetInvalid(row); + } + } +} + static LogicalType FilterlessDebugType() { child_list_t children; children.emplace_back("lower_bound", LogicalType::DOUBLE); @@ -1289,6 +1401,41 @@ static void AddAvgOverloads(AggregateFunctionSet &set, const string &name, aggre finalize, FunctionNullHandling::SPECIAL_HANDLING, FilterlessAvgUpdate, BindFilterlessAvg)); } +template +static AggregateFunction MakeFilterlessExactSumPairFunction(const LogicalType &input_type, + const LogicalType &return_type) { + auto function = AggregateFunction( + "priv_filterless_exact_sum_pair", {input_type, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, + FilterlessLowerPairType(return_type), FilterlessExactSumPairStateSize, FilterlessExactSumPairInitialize, + FilterlessExactSumPairScatterUpdate, FilterlessExactSumPairCombine, FilterlessExactSumPairFinalize, + FunctionNullHandling::SPECIAL_HANDLING, FilterlessExactSumPairUpdate); + function.SetOrderDependent(AggregateOrderDependent::NOT_ORDER_DEPENDENT); + return function; +} + +static unique_ptr BindFilterlessExactDecimalSumPair(ClientContext &, AggregateFunction &function, + vector> &arguments) { + auto input_type = arguments[0]->return_type; + auto return_type = LogicalType::DECIMAL(Decimal::MAX_WIDTH_DECIMAL, DecimalType::GetScale(input_type)); + switch (input_type.InternalType()) { + case PhysicalType::INT16: + function = MakeFilterlessExactSumPairFunction(input_type, return_type); + break; + case PhysicalType::INT32: + function = MakeFilterlessExactSumPairFunction(input_type, return_type); + break; + case PhysicalType::INT64: + function = MakeFilterlessExactSumPairFunction(input_type, return_type); + break; + case PhysicalType::INT128: + function = MakeFilterlessExactSumPairFunction(input_type, return_type); + break; + default: + throw InternalException("priv_filterless_exact_sum_pair: unsupported DECIMAL physical type"); + } + return nullptr; +} + void RegisterFilterlessAggregateFunctions(ExtensionLoader &loader) { auto approx_sum = AggregateFunction::UnaryAggregate( @@ -1324,6 +1471,30 @@ void RegisterFilterlessAggregateFunctions(ExtensionLoader &loader) { count_pair_info.descriptions.push_back(std::move(count_pair_description)); loader.RegisterFunction(std::move(count_pair_info)); + AggregateFunctionSet exact_sum_pair_set("priv_filterless_exact_sum_pair"); + exact_sum_pair_set.AddFunction( + MakeFilterlessExactSumPairFunction(LogicalType::BOOLEAN, LogicalType::HUGEINT)); + exact_sum_pair_set.AddFunction( + MakeFilterlessExactSumPairFunction(LogicalType::TINYINT, LogicalType::HUGEINT)); + exact_sum_pair_set.AddFunction( + MakeFilterlessExactSumPairFunction(LogicalType::SMALLINT, LogicalType::HUGEINT)); + exact_sum_pair_set.AddFunction( + MakeFilterlessExactSumPairFunction(LogicalType::INTEGER, LogicalType::HUGEINT)); + exact_sum_pair_set.AddFunction( + MakeFilterlessExactSumPairFunction(LogicalType::BIGINT, LogicalType::HUGEINT)); + exact_sum_pair_set.AddFunction( + MakeFilterlessExactSumPairFunction(LogicalType::HUGEINT, LogicalType::HUGEINT)); + exact_sum_pair_set.AddFunction(AggregateFunction( + {LogicalTypeId::DECIMAL, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, + FilterlessLowerPairType(LogicalType(LogicalTypeId::DECIMAL)), nullptr, nullptr, nullptr, nullptr, nullptr, + FunctionNullHandling::SPECIAL_HANDLING, nullptr, BindFilterlessExactDecimalSumPair)); + CreateAggregateFunctionInfo exact_sum_pair_info(exact_sum_pair_set); + FunctionDescription exact_sum_pair_description; + exact_sum_pair_description.description = + "[INTERNAL] Fused filtered-answer and sampled-histogram exact SUM partials."; + exact_sum_pair_info.descriptions.push_back(std::move(exact_sum_pair_description)); + loader.RegisterFunction(std::move(exact_sum_pair_info)); + AggregateFunctionSet sum_set("filterless_sum"); AddSumCountOverloads(sum_set, "filterless_sum", FilterlessFinalize, LogicalType::DOUBLE, BindFilterlessSum); diff --git a/src/compiler/privacy_mechanisms.cpp b/src/compiler/privacy_mechanisms.cpp index ffcd312..98cdde2 100644 --- a/src/compiler/privacy_mechanisms.cpp +++ b/src/compiler/privacy_mechanisms.cpp @@ -3166,12 +3166,6 @@ static unique_ptr BuildFilterlessEncodedPuInput(OptimizerExtensionIn return make_uniq(LogicalType::UBIGINT, propagated); } -struct FilterlessLowerComponentInput { - idx_t answer_column; - idx_t histogram_column; - bool paired; -}; - struct FilterlessPreAggregationInput { idx_t group_table_index; idx_t aggregate_table_index; @@ -3180,7 +3174,7 @@ struct FilterlessPreAggregationInput { idx_t active_count_column; vector group_types; vector aggregate_types; - vector components; + vector component_columns; unique_ptr GroupRef(idx_t index) const { return make_uniq(group_types[index], ColumnBinding(group_table_index, index)); @@ -3196,14 +3190,8 @@ struct FilterlessPreAggregationInput { } unique_ptr ComponentRef(OptimizerExtensionInput &input, idx_t index, bool answer) const { - auto &component = components[index]; - auto column = answer ? component.answer_column : component.histogram_column; - auto value = AggregateRef(column); - if (!component.paired) { - return value; - } vector> children; - children.push_back(std::move(value)); + children.push_back(AggregateRef(component_columns[index])); children.push_back(make_uniq(Value(answer ? "answer" : "histogram"))); return BindScalarLocal(input, "struct_extract", std::move(children)); } @@ -3221,16 +3209,16 @@ struct FilterlessPreAggregationInput { } }; -static FilterlessPreAggregationInput -BuildFilterlessPreAggregationInput(const PuPreAggregationInfo &pre, vector components, - idx_t active_count_column) { +static FilterlessPreAggregationInput BuildFilterlessPreAggregationInput(const PuPreAggregationInfo &pre, + vector component_columns, + idx_t active_count_column) { FilterlessPreAggregationInput result; result.group_table_index = pre.lower_agg->group_index; result.aggregate_table_index = pre.lower_agg_index; result.group_count = pre.num_original_groups; result.component_column_offset = 0; result.active_count_column = active_count_column; - result.components = std::move(components); + result.component_columns = std::move(component_columns); for (idx_t i = 0; i < pre.num_original_groups; i++) { result.group_types.push_back(pre.lower_agg->types[i]); } @@ -3242,10 +3230,10 @@ BuildFilterlessPreAggregationInput(const PuPreAggregationInfo &pre, vector components, + vector component_columns, idx_t active_count_column, int64_t max_groups) { if (pre.num_original_groups == 0) { - return BuildFilterlessPreAggregationInput(pre, std::move(components), active_count_column); + return BuildFilterlessPreAggregationInput(pre, std::move(component_columns), active_count_column); } // Flatten the lower aggregate's separate group/aggregate bindings into one projection. The @@ -3296,7 +3284,7 @@ static FilterlessPreAggregationInput ApplyFilterlessMaxGroups(OptimizerExtension result.group_count = pre.num_original_groups; result.component_column_offset = pre.num_original_groups + 1; result.active_count_column = active_count_column; - result.components = std::move(components); + result.component_columns = std::move(component_columns); for (idx_t i = 0; i < pre.num_original_groups; i++) { result.group_types.push_back(pre.lower_agg->types[i]); } @@ -3365,21 +3353,6 @@ static unique_ptr BuildFilterlessLogicalPu(OptimizerExtensionInput & make_uniq(Value::UBIGINT(~uint64_t(1)))); } -static unique_ptr BuildFilterlessExactLowerSumAggregate(OptimizerExtensionInput &input, - const BoundAggregateExpression &aggregate) { - auto input_type = aggregate.children[0]->return_type.InternalType(); - D_ASSERT(input_type != PhysicalType::FLOAT && input_type != PhysicalType::DOUBLE); - return BindPlainAggregate(input, "sum", aggregate.children[0]->Copy()); -} - -static bool CanFuseFilterlessLowerAggregate(const BoundAggregateExpression &aggregate, bool is_count) { - if (is_count) { - return true; - } - auto input_type = aggregate.children[0]->return_type.InternalType(); - return input_type == PhysicalType::FLOAT || input_type == PhysicalType::DOUBLE; -} - static unique_ptr BuildFilterlessLowerPair(OptimizerExtensionInput &input, const BoundAggregateExpression &aggregate, bool is_count, const Expression &encoded_pu, int sample_bits) { @@ -3398,8 +3371,16 @@ static unique_ptr BuildFilterlessLowerPair(OptimizerExtensionInput & } children.push_back(BuildFilterlessIsActive(input, encoded_pu)); children.push_back(BuildFilterlessSamplePredicate(input, encoded_pu, sample_bits)); - return BindAggregateLocal(input, is_count ? "priv_filterless_count_pair" : "priv_filterless_approx_sum_pair", - std::move(children)); + string function_name; + if (is_count) { + function_name = "priv_filterless_count_pair"; + } else { + auto input_type = aggregate.children[0]->return_type.InternalType(); + function_name = input_type == PhysicalType::FLOAT || input_type == PhysicalType::DOUBLE + ? "priv_filterless_approx_sum_pair" + : "priv_filterless_exact_sum_pair"; + } + return BindAggregateLocal(input, function_name, std::move(children)); } void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, OptimizerExtensionInput &input, @@ -3468,32 +3449,17 @@ void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, Optimizer vector count_components; vector> lower_expressions; count_components.reserve(component_count); - lower_expressions.reserve(2 * component_count + 1); - vector lower_components; - lower_components.reserve(component_count); + lower_expressions.reserve(component_count + 1); + vector lower_component_columns; + lower_component_columns.reserve(component_count); for (idx_t i = 0; i < component_count; i++) { auto &aggregate = agg->expressions[i]->Cast(); bool is_count = IsCountAggregate(aggregate); count_components.push_back(is_count); - if (CanFuseFilterlessLowerAggregate(aggregate, is_count)) { - auto column = lower_expressions.size(); - lower_expressions.push_back( - BuildFilterlessLowerPair(input, aggregate, is_count, *encoded_pu, filterless_settings.sample_bits)); - lower_components.push_back({column, column, true}); - continue; - } - - D_ASSERT(!is_count); - auto answer_column = lower_expressions.size(); - auto answer_partial = BuildFilterlessExactLowerSumAggregate(input, aggregate); - answer_partial->Cast().filter = BuildFilterlessIsActive(input, *encoded_pu); - lower_expressions.push_back(std::move(answer_partial)); - auto histogram_column = lower_expressions.size(); - auto histogram_partial = BuildFilterlessExactLowerSumAggregate(input, aggregate); - histogram_partial->Cast().filter = - BuildFilterlessSamplePredicate(input, *encoded_pu, filterless_settings.sample_bits); - lower_expressions.push_back(std::move(histogram_partial)); - lower_components.push_back({answer_column, histogram_column, false}); + auto column = lower_expressions.size(); + lower_expressions.push_back( + BuildFilterlessLowerPair(input, aggregate, is_count, *encoded_pu, filterless_settings.sample_bits)); + lower_component_columns.push_back(column); } idx_t active_count_column = lower_expressions.size(); auto active_count = BindPlainAggregate(input, "count_star", nullptr); @@ -3503,9 +3469,9 @@ void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, Optimizer auto logical_pu = BuildFilterlessLogicalPu(input, std::move(encoded_pu)); auto pre = InsertPuPreAggregation(input, agg, std::move(lower_expressions), std::move(logical_pu)); PRIVACY_DEBUG_PRINT("[dp_filterless] pre-aggregated paired filtered-answer and fixed-sample histogram " - "contributions where supported; floating SUM uses the scalar AS magnitude accumulator"); + "contributions; floating SUM uses the scalar AS magnitude accumulator"); auto pre_input = - ApplyFilterlessMaxGroups(input, agg, pre, std::move(lower_components), active_count_column, max_groups); + ApplyFilterlessMaxGroups(input, agg, pre, std::move(lower_component_columns), active_count_column, max_groups); double visible_cell_epsilon = epsilon / budget_units; for (idx_t i = 0; i < component_count; i++) { double component_epsilon = avg_components.count(i) ? visible_cell_epsilon / 2.0 : visible_cell_epsilon; diff --git a/test/sql/dp_filterless.test b/test/sql/dp_filterless.test index e584827..53708ee 100644 --- a/test/sql/dp_filterless.test +++ b/test/sql/dp_filterless.test @@ -107,6 +107,51 @@ FROM ( ---- 2 2 +query IITT +SELECT x.answer, x.histogram, typeof(x.answer), typeof(x.histogram) +FROM ( + SELECT priv_filterless_exact_sum_pair(value, active, sampled) AS x + FROM (VALUES + (9007199254740993::BIGINT, true, true), + (2::BIGINT, false, true), + (30::BIGINT, true, false), + (NULL::BIGINT, true, true) + ) input(value, active, sampled) +); +---- +9007199254741023 9007199254740995 HUGEINT HUGEINT + +query RRTT +SELECT x.answer, x.histogram, typeof(x.answer), typeof(x.histogram) +FROM ( + SELECT priv_filterless_exact_sum_pair(value, active, sampled) AS x + FROM (VALUES + (90071992547409.93::DECIMAL(38, 2), true, true), + (0.02::DECIMAL(38, 2), false, true), + (0.30::DECIMAL(38, 2), true, false), + (NULL::DECIMAL(38, 2), true, true) + ) input(value, active, sampled) +); +---- +90071992547410.23 90071992547409.95 DECIMAL(38,2) DECIMAL(38,2) + +# Every exact SUM input accepted by filterless binds to an exact HUGEINT +# partial; DECIMAL is bound separately above to retain its scale. +query TTTTT +SELECT typeof(b.answer), typeof(t.answer), typeof(s.answer), typeof(i.answer), typeof(h.answer) +FROM ( + SELECT priv_filterless_exact_sum_pair(b, true, true) AS b, + priv_filterless_exact_sum_pair(t, true, true) AS t, + priv_filterless_exact_sum_pair(s, true, true) AS s, + priv_filterless_exact_sum_pair(i, true, true) AS i, + priv_filterless_exact_sum_pair(h, true, true) AS h + FROM (VALUES + (true::BOOLEAN, 1::TINYINT, 1::SMALLINT, 1::INTEGER, 1::HUGEINT) + ) input(b, t, s, i, h) +); +---- +HUGEINT HUGEINT HUGEINT HUGEINT HUGEINT + # Sampling is deterministic from the high PU bits. At p=6, PU 0 is sampled and # receives weight 64; PUs with the high bit set are not sampled. Qualifying # non-sampled PUs enter the answer but not the histogram. @@ -531,8 +576,8 @@ WHERE NOT qualifies; ---- 2000.000000 2 1000.000000 -# The compiler uses one paired lower aggregate per floating SUM/COUNT -# component, instead of separate filtered and sampled aggregates. +# The compiler uses one paired lower aggregate per SUM/COUNT component, +# instead of separate filtered and sampled aggregates. statement ok SET explain_output = 'physical_only'; @@ -605,6 +650,12 @@ INSERT INTO filterless_exact_numeric VALUES (1, 9007199254740993, 90071992547409.93), (2, 2, 0.02); +query II +EXPLAIN SELECT SUM(integer_amount), SUM(decimal_amount), COUNT(*) +FROM filterless_exact_numeric; +---- +physical_plan :[\s\S]*priv_filterless_exact_sum_p[\s\S]*air[\s\S]*priv_filterless_count_pair[\s\S]* + query IRTT SELECT SUM(integer_amount), SUM(decimal_amount), typeof(SUM(integer_amount)), typeof(SUM(decimal_amount)) From c57da501b28859377da21ca905c3a13d67d38c16 Mon Sep 17 00:00:00 2001 From: ila Date: Tue, 1 Sep 2026 13:13:02 +0200 Subject: [PATCH 4/6] Simplify filterless partial fusion --- src/aggregates/filterless_aggregate.cpp | 468 ++++++++---------------- src/compiler/privacy_mechanisms.cpp | 52 +-- test/sql/dp_filterless.test | 76 +--- 3 files changed, 167 insertions(+), 429 deletions(-) diff --git a/src/aggregates/filterless_aggregate.cpp b/src/aggregates/filterless_aggregate.cpp index 13252fe..220eee1 100644 --- a/src/aggregates/filterless_aggregate.cpp +++ b/src/aggregates/filterless_aggregate.cpp @@ -482,6 +482,10 @@ static hugeint_t ToHugeint(hugeint_t value) { return value; } +static hugeint_t ToHugeint(bool value) { + return Hugeint::Convert(static_cast(value)); +} + template static hugeint_t ToHugeint(INPUT_TYPE value) { return Hugeint::Convert(value); @@ -915,30 +919,24 @@ static void FilterlessExactFinalize(Vector &states, AggregateInputData &input, V } } -// Internal scalar form of the as_clip_sum magnitude accumulator. The compiler -// uses it for the per-PU floating SUM below the filterless aggregate, so an -// unstable ordinary DOUBLE SUM cannot erase small contributions before the -// contribution bound is applied. struct FilterlessApproxSumState { bool isset; hugeint_t positive; hugeint_t negative; }; -struct FilterlessApproxSumOperation { - template - static void Initialize(STATE &state) { - state.isset = false; - state.positive = hugeint_t(0); - state.negative = hugeint_t(0); - } +// Use the same scaled-magnitude representation as PAC's approximate SUM so +// cancellation cannot erase small per-PU contributions before clipping. +struct FilterlessApproxSumPairOperation { + using INPUT_TYPE = double; + using STATE = FilterlessApproxSumState; + using RESULT_TYPE = double; static hugeint_t ApproximateScaledValue(double value) { return Hugeint::Convert(ClipApproximateMagnitude64(AsScaledMagnitude(value))); } - template - static void AddValue(STATE &state, double value) { + static void Add(STATE &state, INPUT_TYPE value) { if (!std::isfinite(value)) { throw InvalidInputException("filterless: per-PU SUM contribution must be finite"); } @@ -951,127 +949,101 @@ struct FilterlessApproxSumOperation { } } - template - static void Operation(STATE &state, const INPUT_TYPE &input, AggregateUnaryInput &) { - AddValue(state, static_cast(input)); - } - - template - static void ConstantOperation(STATE &state, const INPUT_TYPE &input, AggregateUnaryInput &, idx_t count) { - double value = static_cast(input); - if (!std::isfinite(value)) { - throw InvalidInputException("filterless: per-PU SUM contribution must be finite"); - } - state.isset = true; - auto total = Hugeint::Multiply(ApproximateScaledValue(value), Hugeint::Convert(count)); - if (std::signbit(value) && value != 0.0) { - state.negative = Hugeint::Add(state.negative, total); - } else { - state.positive = Hugeint::Add(state.positive, total); - } - } - - template - static void Combine(const STATE &source, STATE &target, AggregateInputData &) { + static void Combine(const STATE &source, STATE &target) { target.isset = target.isset || source.isset; target.positive = Hugeint::Add(target.positive, source.positive); target.negative = Hugeint::Add(target.negative, source.negative); } - template - static double FinalizeValue(const STATE &state) { - auto scaled = Hugeint::Subtract(state.positive, state.negative); - return Hugeint::Cast(scaled) / CLIP_DOUBLE_SCALE; + static bool IsSet(const STATE &state) { + return state.isset; } - template - static void Finalize(STATE &state, RESULT_TYPE &target, AggregateFinalizeData &finalize_data) { - if (!state.isset) { - finalize_data.ReturnNull(); - return; - } - target = FinalizeValue(state); - } - - static bool IgnoreNull() { - return true; + static RESULT_TYPE Finalize(const STATE &state) { + auto scaled = Hugeint::Subtract(state.positive, state.negative); + return Hugeint::Cast(scaled) / CLIP_DOUBLE_SCALE; } }; -struct FilterlessApproxSumPairState { - FilterlessApproxSumState answer; - FilterlessApproxSumState histogram; +struct FilterlessExactSumPartialState { + bool isset; + hugeint_t value; }; -struct FilterlessCountPairState { - uint64_t answer; - uint64_t histogram; +template +struct FilterlessLowerPairState { + PARTIAL_STATE answer; + PARTIAL_STATE histogram; }; -struct FilterlessExactSumPartialState { - bool isset; - hugeint_t value; +struct FilterlessCountPairOperation { + using INPUT_TYPE = bool; + using STATE = uint64_t; + using RESULT_TYPE = int64_t; - void Initialize() { - isset = false; - value = hugeint_t(0); + static void Add(STATE &state, INPUT_TYPE value) { + state += static_cast(value); } - void Combine(const FilterlessExactSumPartialState &other) { - isset = isset || other.isset; - value = Hugeint::Add(value, other.value); + static void Combine(const STATE &source, STATE &target) { + target += source; + } + + static bool IsSet(const STATE &) { + return true; } -}; -struct FilterlessExactSumPairState { - FilterlessExactSumPartialState answer; - FilterlessExactSumPartialState histogram; + static RESULT_TYPE Finalize(const STATE &state) { + return static_cast(state); + } }; -static LogicalType FilterlessLowerPairType(const LogicalType &value_type) { - child_list_t children; - children.emplace_back("answer", value_type); - children.emplace_back("histogram", value_type); - return LogicalType::STRUCT(std::move(children)); -} +template +struct FilterlessExactSumPairOperation { + using INPUT_TYPE = INPUT; + using STATE = FilterlessExactSumPartialState; + using RESULT_TYPE = hugeint_t; -static idx_t FilterlessApproxSumPairStateSize(const AggregateFunction &) { - return sizeof(FilterlessApproxSumPairState); -} + static void Add(STATE &state, INPUT_TYPE value) { + state.isset = true; + state.value = Hugeint::Add(state.value, ToHugeint(value)); + } -static idx_t FilterlessCountPairStateSize(const AggregateFunction &) { - return sizeof(FilterlessCountPairState); -} + static void Combine(const STATE &source, STATE &target) { + target.isset = target.isset || source.isset; + target.value = Hugeint::Add(target.value, source.value); + } -static idx_t FilterlessExactSumPairStateSize(const AggregateFunction &) { - return sizeof(FilterlessExactSumPairState); -} + static bool IsSet(const STATE &state) { + return state.isset; + } -static void FilterlessApproxSumPairInitialize(const AggregateFunction &, data_ptr_t state_p) { - auto &state = *reinterpret_cast(state_p); - FilterlessApproxSumOperation::Initialize(state.answer); - FilterlessApproxSumOperation::Initialize(state.histogram); + static RESULT_TYPE Finalize(const STATE &state) { + return state.value; + } +}; + +static LogicalType FilterlessLowerPairType(const LogicalType &value_type) { + return LogicalType::STRUCT({{"answer", value_type}, {"histogram", value_type}}); } -static void FilterlessCountPairInitialize(const AggregateFunction &, data_ptr_t state_p) { - memset(state_p, 0, sizeof(FilterlessCountPairState)); +template +static idx_t FilterlessLowerPairStateSize(const AggregateFunction &) { + return sizeof(FilterlessLowerPairState); } -static void FilterlessExactSumPairInitialize(const AggregateFunction &, data_ptr_t state_p) { - auto &state = *reinterpret_cast(state_p); - state.answer.Initialize(); - state.histogram.Initialize(); +template +static void FilterlessLowerPairInitialize(const AggregateFunction &, data_ptr_t state_p) { + memset(state_p, 0, sizeof(FilterlessLowerPairState)); } -template -static void FilterlessLowerPairUpdateRows(Vector inputs[], idx_t count, STATE_GETTER get_state, UPDATE update) { - UnifiedVectorFormat value_data; - UnifiedVectorFormat active_data; - UnifiedVectorFormat sampled_data; +template +static void FilterlessLowerPairUpdateRows(Vector inputs[], idx_t count, STATE_GETTER get_state) { + UnifiedVectorFormat value_data, active_data, sampled_data; inputs[0].ToUnifiedFormat(count, value_data); inputs[1].ToUnifiedFormat(count, active_data); inputs[2].ToUnifiedFormat(count, sampled_data); - auto values = UnifiedVectorFormat::GetData(value_data); + auto values = UnifiedVectorFormat::GetData(value_data); auto active = UnifiedVectorFormat::GetData(active_data); auto sampled = UnifiedVectorFormat::GetData(sampled_data); for (idx_t row = 0; row < count; row++) { @@ -1083,193 +1055,62 @@ static void FilterlessLowerPairUpdateRows(Vector inputs[], idx_t count, STATE_GE auto sampled_index = sampled_data.sel->get_index(row); bool is_active = active_data.validity.RowIsValid(active_index) && active[active_index]; bool is_sampled = sampled_data.validity.RowIsValid(sampled_index) && sampled[sampled_index]; - if (!is_active && !is_sampled) { - continue; + if (is_active || is_sampled) { + auto state = get_state(row); + if (is_active) { + OPERATION::Add(state->answer, values[value_index]); + } + if (is_sampled) { + OPERATION::Add(state->histogram, values[value_index]); + } } - update(*get_state(row), values[value_index], is_active, is_sampled); } } -static void UpdateFilterlessApproxSumPair(FilterlessApproxSumPairState &state, double value, bool active, - bool sampled) { - if (active) { - FilterlessApproxSumOperation::AddValue(state.answer, value); - } - if (sampled) { - FilterlessApproxSumOperation::AddValue(state.histogram, value); - } +template +static void FilterlessLowerPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, idx_t count) { + auto state = reinterpret_cast *>(state_p); + FilterlessLowerPairUpdateRows(inputs, count, [state](idx_t) { return state; }); } -static void UpdateFilterlessCountPair(FilterlessCountPairState &state, bool count_value, bool active, bool sampled) { - if (!count_value) { - return; - } - state.answer += static_cast(active); - state.histogram += static_cast(sampled); -} - -template -static void AddFilterlessExactInteger(hugeint_t &result, INPUT_TYPE value) { - // Match DuckDB's exact integer SUM accumulator: add the two's-complement low - // word and adjust the high word only on carry or borrow. - auto lower = static_cast(value); - result.lower += lower; - int overflow = result.lower < lower; - int positive = value >= 0; - if (!(overflow ^ positive)) { - result.upper += -1 + 2 * positive; - } -} - -template <> -void AddFilterlessExactInteger(hugeint_t &result, hugeint_t value) { - result = Hugeint::Add(result, value); -} - -template -static void AddFilterlessExactSumPairValue(FilterlessExactSumPairState &state, INPUT_TYPE value, bool active, - bool sampled) { - if (active) { - state.answer.isset = true; - AddFilterlessExactInteger(state.answer.value, value); - } - if (sampled) { - state.histogram.isset = true; - AddFilterlessExactInteger(state.histogram.value, value); - } -} - -template -static void UpdateFilterlessExactSumPair(FilterlessExactSumPairState &state, INPUT_TYPE value, bool active, - bool sampled) { - AddFilterlessExactSumPairValue(state, value, active, sampled); -} - -static void FilterlessApproxSumPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, - idx_t count) { - auto state = reinterpret_cast(state_p); - FilterlessLowerPairUpdateRows( - inputs, count, [state](idx_t) { return state; }, UpdateFilterlessApproxSumPair); -} - -static void FilterlessCountPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, idx_t count) { - auto state = reinterpret_cast(state_p); - FilterlessLowerPairUpdateRows( - inputs, count, [state](idx_t) { return state; }, UpdateFilterlessCountPair); -} - -template -static void FilterlessExactSumPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, - idx_t count) { - auto state = reinterpret_cast(state_p); - FilterlessLowerPairUpdateRows( - inputs, count, [state](idx_t) { return state; }, UpdateFilterlessExactSumPair); -} - -template -static void FilterlessLowerPairScatterUpdate(Vector inputs[], Vector &states, idx_t count, UPDATE update) { +template +static void FilterlessLowerPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, + idx_t count) { UnifiedVectorFormat state_data; states.ToUnifiedFormat(count, state_data); - auto state_ptrs = UnifiedVectorFormat::GetData(state_data); - FilterlessLowerPairUpdateRows( - inputs, count, [&](idx_t row) { return state_ptrs[state_data.sel->get_index(row)]; }, update); -} - -static void FilterlessApproxSumPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, - idx_t count) { - FilterlessLowerPairScatterUpdate(inputs, states, count, - UpdateFilterlessApproxSumPair); + auto state_ptrs = UnifiedVectorFormat::GetData *>(state_data); + FilterlessLowerPairUpdateRows(inputs, count, + [&](idx_t row) { return state_ptrs[state_data.sel->get_index(row)]; }); } -static void FilterlessCountPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, - idx_t count) { - FilterlessLowerPairScatterUpdate(inputs, states, count, UpdateFilterlessCountPair); -} - -template -static void FilterlessExactSumPairScatterUpdate(Vector inputs[], AggregateInputData &, idx_t, Vector &states, - idx_t count) { - FilterlessLowerPairScatterUpdate(inputs, states, count, - UpdateFilterlessExactSumPair); -} - -static void FilterlessApproxSumPairCombine(Vector &source, Vector &target, AggregateInputData &input, idx_t count) { - auto sources = FlatVector::GetData(source); - auto targets = FlatVector::GetData(target); +template +static void FilterlessLowerPairCombine(Vector &source, Vector &target, AggregateInputData &, idx_t count) { + using PAIR_STATE = FilterlessLowerPairState; + auto sources = FlatVector::GetData(source); + auto targets = FlatVector::GetData(target); for (idx_t i = 0; i < count; i++) { - FilterlessApproxSumOperation::Combine( - sources[i]->answer, targets[i]->answer, input); - FilterlessApproxSumOperation::Combine( - sources[i]->histogram, targets[i]->histogram, input); + OPERATION::Combine(sources[i]->answer, targets[i]->answer); + OPERATION::Combine(sources[i]->histogram, targets[i]->histogram); } } -static void FilterlessCountPairCombine(Vector &source, Vector &target, AggregateInputData &, idx_t count) { - auto sources = FlatVector::GetData(source); - auto targets = FlatVector::GetData(target); - for (idx_t i = 0; i < count; i++) { - targets[i]->answer += sources[i]->answer; - targets[i]->histogram += sources[i]->histogram; - } -} - -static void FilterlessExactSumPairCombine(Vector &source, Vector &target, AggregateInputData &, idx_t count) { - auto sources = FlatVector::GetData(source); - auto targets = FlatVector::GetData(target); - for (idx_t i = 0; i < count; i++) { - targets[i]->answer.Combine(sources[i]->answer); - targets[i]->histogram.Combine(sources[i]->histogram); - } -} - -static void FilterlessApproxSumPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, - idx_t offset) { - auto state_ptrs = FlatVector::GetData(states); - auto &children = StructVector::GetEntries(result); - auto answers = FlatVector::GetData(*children[0]); - auto histograms = FlatVector::GetData(*children[1]); - for (idx_t i = 0; i < count; i++) { - auto row = offset + i; - if (state_ptrs[i]->answer.isset) { - answers[row] = FilterlessApproxSumOperation::FinalizeValue(state_ptrs[i]->answer); - } else { - FlatVector::Validity(*children[0]).SetInvalid(row); - } - if (state_ptrs[i]->histogram.isset) { - histograms[row] = FilterlessApproxSumOperation::FinalizeValue(state_ptrs[i]->histogram); - } else { - FlatVector::Validity(*children[1]).SetInvalid(row); - } - } -} - -static void FilterlessCountPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, +template +static void FilterlessLowerPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, idx_t offset) { - auto state_ptrs = FlatVector::GetData(states); + using PAIR_STATE = FilterlessLowerPairState; + auto state_ptrs = FlatVector::GetData(states); auto &children = StructVector::GetEntries(result); - auto answers = FlatVector::GetData(*children[0]); - auto histograms = FlatVector::GetData(*children[1]); - for (idx_t i = 0; i < count; i++) { - answers[offset + i] = static_cast(state_ptrs[i]->answer); - histograms[offset + i] = static_cast(state_ptrs[i]->histogram); - } -} - -static void FilterlessExactSumPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, - idx_t offset) { - auto state_ptrs = FlatVector::GetData(states); - auto &children = StructVector::GetEntries(result); - auto answers = FlatVector::GetData(*children[0]); - auto histograms = FlatVector::GetData(*children[1]); + auto answers = FlatVector::GetData(*children[0]); + auto histograms = FlatVector::GetData(*children[1]); for (idx_t i = 0; i < count; i++) { auto row = offset + i; - if (state_ptrs[i]->answer.isset) { - answers[row] = state_ptrs[i]->answer.value; + if (OPERATION::IsSet(state_ptrs[i]->answer)) { + answers[row] = OPERATION::Finalize(state_ptrs[i]->answer); } else { FlatVector::Validity(*children[0]).SetInvalid(row); } - if (state_ptrs[i]->histogram.isset) { - histograms[row] = state_ptrs[i]->histogram.value; + if (OPERATION::IsSet(state_ptrs[i]->histogram)) { + histograms[row] = OPERATION::Finalize(state_ptrs[i]->histogram); } else { FlatVector::Validity(*children[1]).SetInvalid(row); } @@ -1401,20 +1242,28 @@ static void AddAvgOverloads(AggregateFunctionSet &set, const string &name, aggre finalize, FunctionNullHandling::SPECIAL_HANDLING, FilterlessAvgUpdate, BindFilterlessAvg)); } +template +static AggregateFunction MakeFilterlessLowerPairFunction(const string &name, const LogicalType &input_type, + const LogicalType &return_type) { + auto function = + AggregateFunction(name, {input_type, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, + FilterlessLowerPairType(return_type), FilterlessLowerPairStateSize, + FilterlessLowerPairInitialize, FilterlessLowerPairScatterUpdate, + FilterlessLowerPairCombine, FilterlessLowerPairFinalize, + FunctionNullHandling::SPECIAL_HANDLING, FilterlessLowerPairUpdate); + function.SetOrderDependent(AggregateOrderDependent::NOT_ORDER_DEPENDENT); + return function; +} + template static AggregateFunction MakeFilterlessExactSumPairFunction(const LogicalType &input_type, const LogicalType &return_type) { - auto function = AggregateFunction( - "priv_filterless_exact_sum_pair", {input_type, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, - FilterlessLowerPairType(return_type), FilterlessExactSumPairStateSize, FilterlessExactSumPairInitialize, - FilterlessExactSumPairScatterUpdate, FilterlessExactSumPairCombine, FilterlessExactSumPairFinalize, - FunctionNullHandling::SPECIAL_HANDLING, FilterlessExactSumPairUpdate); - function.SetOrderDependent(AggregateOrderDependent::NOT_ORDER_DEPENDENT); - return function; + return MakeFilterlessLowerPairFunction>("priv_filterless_sum_pair", + input_type, return_type); } -static unique_ptr BindFilterlessExactDecimalSumPair(ClientContext &, AggregateFunction &function, - vector> &arguments) { +static unique_ptr BindFilterlessDecimalSumPair(ClientContext &, AggregateFunction &function, + vector> &arguments) { auto input_type = arguments[0]->return_type; auto return_type = LogicalType::DECIMAL(Decimal::MAX_WIDTH_DECIMAL, DecimalType::GetScale(input_type)); switch (input_type.InternalType()) { @@ -1431,69 +1280,38 @@ static unique_ptr BindFilterlessExactDecimalSumPair(ClientContext function = MakeFilterlessExactSumPairFunction(input_type, return_type); break; default: - throw InternalException("priv_filterless_exact_sum_pair: unsupported DECIMAL physical type"); + throw InternalException("priv_filterless_sum_pair: unsupported DECIMAL physical type"); } return nullptr; } void RegisterFilterlessAggregateFunctions(ExtensionLoader &loader) { - auto approx_sum = - AggregateFunction::UnaryAggregate( - LogicalType::DOUBLE, LogicalType::DOUBLE); - approx_sum.name = "filterless_approx_sum"; - CreateAggregateFunctionInfo approx_sum_info(approx_sum); - FunctionDescription approx_sum_description; - approx_sum_description.description = - "[INTERNAL] Scalar AS magnitude sum used by dp_filterless per-PU pre-aggregation."; - approx_sum_info.descriptions.push_back(std::move(approx_sum_description)); - loader.RegisterFunction(std::move(approx_sum_info)); - - AggregateFunction approx_sum_pair( - "priv_filterless_approx_sum_pair", {LogicalType::DOUBLE, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, - FilterlessLowerPairType(LogicalType::DOUBLE), FilterlessApproxSumPairStateSize, - FilterlessApproxSumPairInitialize, FilterlessApproxSumPairScatterUpdate, FilterlessApproxSumPairCombine, - FilterlessApproxSumPairFinalize, FunctionNullHandling::SPECIAL_HANDLING, FilterlessApproxSumPairUpdate); - CreateAggregateFunctionInfo approx_sum_pair_info(approx_sum_pair); - FunctionDescription approx_sum_pair_description; - approx_sum_pair_description.description = - "[INTERNAL] Fused filtered-answer and sampled-histogram approximate SUM partials."; - approx_sum_pair_info.descriptions.push_back(std::move(approx_sum_pair_description)); - loader.RegisterFunction(std::move(approx_sum_pair_info)); - - AggregateFunction count_pair( - "priv_filterless_count_pair", {LogicalType::BOOLEAN, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, - FilterlessLowerPairType(LogicalType::BIGINT), FilterlessCountPairStateSize, FilterlessCountPairInitialize, - FilterlessCountPairScatterUpdate, FilterlessCountPairCombine, FilterlessCountPairFinalize, - FunctionNullHandling::SPECIAL_HANDLING, FilterlessCountPairUpdate); + AggregateFunction count_pair = MakeFilterlessLowerPairFunction( + "priv_filterless_count_pair", LogicalType::BOOLEAN, LogicalType::BIGINT); CreateAggregateFunctionInfo count_pair_info(count_pair); FunctionDescription count_pair_description; count_pair_description.description = "[INTERNAL] Fused filtered-answer and sampled-histogram COUNT partials."; count_pair_info.descriptions.push_back(std::move(count_pair_description)); loader.RegisterFunction(std::move(count_pair_info)); - AggregateFunctionSet exact_sum_pair_set("priv_filterless_exact_sum_pair"); - exact_sum_pair_set.AddFunction( - MakeFilterlessExactSumPairFunction(LogicalType::BOOLEAN, LogicalType::HUGEINT)); - exact_sum_pair_set.AddFunction( - MakeFilterlessExactSumPairFunction(LogicalType::TINYINT, LogicalType::HUGEINT)); - exact_sum_pair_set.AddFunction( - MakeFilterlessExactSumPairFunction(LogicalType::SMALLINT, LogicalType::HUGEINT)); - exact_sum_pair_set.AddFunction( - MakeFilterlessExactSumPairFunction(LogicalType::INTEGER, LogicalType::HUGEINT)); - exact_sum_pair_set.AddFunction( - MakeFilterlessExactSumPairFunction(LogicalType::BIGINT, LogicalType::HUGEINT)); - exact_sum_pair_set.AddFunction( - MakeFilterlessExactSumPairFunction(LogicalType::HUGEINT, LogicalType::HUGEINT)); - exact_sum_pair_set.AddFunction(AggregateFunction( + AggregateFunctionSet sum_pair_set("priv_filterless_sum_pair"); + sum_pair_set.AddFunction(MakeFilterlessLowerPairFunction( + "priv_filterless_sum_pair", LogicalType::DOUBLE, LogicalType::DOUBLE)); + sum_pair_set.AddFunction(MakeFilterlessExactSumPairFunction(LogicalType::BOOLEAN, LogicalType::HUGEINT)); + sum_pair_set.AddFunction(MakeFilterlessExactSumPairFunction(LogicalType::TINYINT, LogicalType::HUGEINT)); + sum_pair_set.AddFunction(MakeFilterlessExactSumPairFunction(LogicalType::SMALLINT, LogicalType::HUGEINT)); + sum_pair_set.AddFunction(MakeFilterlessExactSumPairFunction(LogicalType::INTEGER, LogicalType::HUGEINT)); + sum_pair_set.AddFunction(MakeFilterlessExactSumPairFunction(LogicalType::BIGINT, LogicalType::HUGEINT)); + sum_pair_set.AddFunction(MakeFilterlessExactSumPairFunction(LogicalType::HUGEINT, LogicalType::HUGEINT)); + sum_pair_set.AddFunction(AggregateFunction( {LogicalTypeId::DECIMAL, LogicalType::BOOLEAN, LogicalType::BOOLEAN}, FilterlessLowerPairType(LogicalType(LogicalTypeId::DECIMAL)), nullptr, nullptr, nullptr, nullptr, nullptr, - FunctionNullHandling::SPECIAL_HANDLING, nullptr, BindFilterlessExactDecimalSumPair)); - CreateAggregateFunctionInfo exact_sum_pair_info(exact_sum_pair_set); - FunctionDescription exact_sum_pair_description; - exact_sum_pair_description.description = - "[INTERNAL] Fused filtered-answer and sampled-histogram exact SUM partials."; - exact_sum_pair_info.descriptions.push_back(std::move(exact_sum_pair_description)); - loader.RegisterFunction(std::move(exact_sum_pair_info)); + FunctionNullHandling::SPECIAL_HANDLING, nullptr, BindFilterlessDecimalSumPair)); + CreateAggregateFunctionInfo sum_pair_info(sum_pair_set); + FunctionDescription sum_pair_description; + sum_pair_description.description = "[INTERNAL] Fused filtered-answer and sampled-histogram SUM partials."; + sum_pair_info.descriptions.push_back(std::move(sum_pair_description)); + loader.RegisterFunction(std::move(sum_pair_info)); AggregateFunctionSet sum_set("filterless_sum"); AddSumCountOverloads(sum_set, "filterless_sum", FilterlessFinalize, LogicalType::DOUBLE, diff --git a/src/compiler/privacy_mechanisms.cpp b/src/compiler/privacy_mechanisms.cpp index 98cdde2..7786ea1 100644 --- a/src/compiler/privacy_mechanisms.cpp +++ b/src/compiler/privacy_mechanisms.cpp @@ -3171,10 +3171,9 @@ struct FilterlessPreAggregationInput { idx_t aggregate_table_index; idx_t group_count; idx_t component_column_offset; - idx_t active_count_column; + idx_t component_count; vector group_types; vector aggregate_types; - vector component_columns; unique_ptr GroupRef(idx_t index) const { return make_uniq(group_types[index], ColumnBinding(group_table_index, index)); @@ -3191,7 +3190,7 @@ struct FilterlessPreAggregationInput { unique_ptr ComponentRef(OptimizerExtensionInput &input, idx_t index, bool answer) const { vector> children; - children.push_back(AggregateRef(component_columns[index])); + children.push_back(AggregateRef(index)); children.push_back(make_uniq(Value(answer ? "answer" : "histogram"))); return BindScalarLocal(input, "struct_extract", std::move(children)); } @@ -3205,42 +3204,39 @@ struct FilterlessPreAggregationInput { } unique_ptr ActiveCountRef() const { - return AggregateRef(active_count_column); + return AggregateRef(component_count); } }; static FilterlessPreAggregationInput BuildFilterlessPreAggregationInput(const PuPreAggregationInfo &pre, - vector component_columns, - idx_t active_count_column) { + idx_t component_count) { FilterlessPreAggregationInput result; result.group_table_index = pre.lower_agg->group_index; result.aggregate_table_index = pre.lower_agg_index; result.group_count = pre.num_original_groups; result.component_column_offset = 0; - result.active_count_column = active_count_column; - result.component_columns = std::move(component_columns); + result.component_count = component_count; for (idx_t i = 0; i < pre.num_original_groups; i++) { result.group_types.push_back(pre.lower_agg->types[i]); } - for (idx_t i = 0; i < pre.lower_agg->expressions.size(); i++) { + for (idx_t i = 0; i < component_count + 1; i++) { result.aggregate_types.push_back(pre.lower_agg->types[pre.num_original_groups + 1 + i]); } return result; } static FilterlessPreAggregationInput ApplyFilterlessMaxGroups(OptimizerExtensionInput &input, LogicalAggregate *agg, - const PuPreAggregationInfo &pre, - vector component_columns, - idx_t active_count_column, int64_t max_groups) { + const PuPreAggregationInfo &pre, idx_t component_count, + int64_t max_groups) { if (pre.num_original_groups == 0) { - return BuildFilterlessPreAggregationInput(pre, std::move(component_columns), active_count_column); + return BuildFilterlessPreAggregationInput(pre, component_count); } // Flatten the lower aggregate's separate group/aggregate bindings into one projection. The // lower aggregate already emits exactly one row per logical (PU, SQL group), so ROW_NUMBER // caps the group set without needing duplicate-aware DENSE_RANK. DuckDB can optimize this // single hashed ordering into a bounded per-PU top-k aggregate. - idx_t lower_aggregate_count = pre.lower_agg->expressions.size(); + idx_t lower_aggregate_count = component_count + 1; idx_t projection_index = input.optimizer.binder.GenerateTableIndex(); vector> expressions; expressions.reserve(pre.num_original_groups + 1 + lower_aggregate_count); @@ -3283,8 +3279,7 @@ static FilterlessPreAggregationInput ApplyFilterlessMaxGroups(OptimizerExtension result.aggregate_table_index = projection_index; result.group_count = pre.num_original_groups; result.component_column_offset = pre.num_original_groups + 1; - result.active_count_column = active_count_column; - result.component_columns = std::move(component_columns); + result.component_count = component_count; for (idx_t i = 0; i < pre.num_original_groups; i++) { result.group_types.push_back(pre.lower_agg->types[i]); } @@ -3359,10 +3354,7 @@ static unique_ptr BuildFilterlessLowerPair(OptimizerExtensionInput & vector> children; if (is_count) { if (aggregate.function.name == "count" && !aggregate.children.empty()) { - auto countable = - make_uniq(ExpressionType::OPERATOR_IS_NOT_NULL, LogicalType::BOOLEAN); - countable->children.push_back(aggregate.children[0]->Copy()); - children.push_back(std::move(countable)); + children.push_back(BuildIsNotNullPredicate(aggregate.children[0]->Copy())); } else { children.push_back(make_uniq(Value::BOOLEAN(true))); } @@ -3371,16 +3363,8 @@ static unique_ptr BuildFilterlessLowerPair(OptimizerExtensionInput & } children.push_back(BuildFilterlessIsActive(input, encoded_pu)); children.push_back(BuildFilterlessSamplePredicate(input, encoded_pu, sample_bits)); - string function_name; - if (is_count) { - function_name = "priv_filterless_count_pair"; - } else { - auto input_type = aggregate.children[0]->return_type.InternalType(); - function_name = input_type == PhysicalType::FLOAT || input_type == PhysicalType::DOUBLE - ? "priv_filterless_approx_sum_pair" - : "priv_filterless_exact_sum_pair"; - } - return BindAggregateLocal(input, function_name, std::move(children)); + return BindAggregateLocal(input, is_count ? "priv_filterless_count_pair" : "priv_filterless_sum_pair", + std::move(children)); } void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, OptimizerExtensionInput &input, @@ -3450,18 +3434,13 @@ void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, Optimizer vector> lower_expressions; count_components.reserve(component_count); lower_expressions.reserve(component_count + 1); - vector lower_component_columns; - lower_component_columns.reserve(component_count); for (idx_t i = 0; i < component_count; i++) { auto &aggregate = agg->expressions[i]->Cast(); bool is_count = IsCountAggregate(aggregate); count_components.push_back(is_count); - auto column = lower_expressions.size(); lower_expressions.push_back( BuildFilterlessLowerPair(input, aggregate, is_count, *encoded_pu, filterless_settings.sample_bits)); - lower_component_columns.push_back(column); } - idx_t active_count_column = lower_expressions.size(); auto active_count = BindPlainAggregate(input, "count_star", nullptr); active_count->Cast().filter = BuildFilterlessIsActive(input, *encoded_pu); lower_expressions.push_back(std::move(active_count)); @@ -3470,8 +3449,7 @@ void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, Optimizer auto pre = InsertPuPreAggregation(input, agg, std::move(lower_expressions), std::move(logical_pu)); PRIVACY_DEBUG_PRINT("[dp_filterless] pre-aggregated paired filtered-answer and fixed-sample histogram " "contributions; floating SUM uses the scalar AS magnitude accumulator"); - auto pre_input = - ApplyFilterlessMaxGroups(input, agg, pre, std::move(lower_component_columns), active_count_column, max_groups); + auto pre_input = ApplyFilterlessMaxGroups(input, agg, pre, component_count, max_groups); double visible_cell_epsilon = epsilon / budget_units; for (idx_t i = 0; i < component_count; i++) { double component_epsilon = avg_components.count(i) ? visible_cell_epsilon / 2.0 : visible_cell_epsilon; diff --git a/test/sql/dp_filterless.test b/test/sql/dp_filterless.test index 53708ee..5df7ee9 100644 --- a/test/sql/dp_filterless.test +++ b/test/sql/dp_filterless.test @@ -77,74 +77,16 @@ FROM ( ---- 25.000000 100.000000 4.000000 -# The internal lower aggregates compute filtered-answer and sampled-histogram -# partials together. NULL values retain SUM and COUNT's ordinary semantics. -query RR -SELECT x.answer, x.histogram -FROM ( - SELECT priv_filterless_approx_sum_pair(value, active, sampled) AS x - FROM (VALUES - (10.0::DOUBLE, true, true), - (20.0::DOUBLE, false, true), - (NULL::DOUBLE, true, true), - (30.0::DOUBLE, true, false) - ) input(value, active, sampled) -); ----- -40.000000 30.000000 - -query II -SELECT x.answer, x.histogram -FROM ( - SELECT priv_filterless_count_pair(countable, active, sampled) AS x - FROM (VALUES - (true, true, true), - (false, true, true), - (true, false, true), - (true, true, false) - ) input(countable, active, sampled) -); ----- -2 2 - -query IITT -SELECT x.answer, x.histogram, typeof(x.answer), typeof(x.histogram) -FROM ( - SELECT priv_filterless_exact_sum_pair(value, active, sampled) AS x - FROM (VALUES - (9007199254740993::BIGINT, true, true), - (2::BIGINT, false, true), - (30::BIGINT, true, false), - (NULL::BIGINT, true, true) - ) input(value, active, sampled) -); ----- -9007199254741023 9007199254740995 HUGEINT HUGEINT - -query RRTT -SELECT x.answer, x.histogram, typeof(x.answer), typeof(x.histogram) -FROM ( - SELECT priv_filterless_exact_sum_pair(value, active, sampled) AS x - FROM (VALUES - (90071992547409.93::DECIMAL(38, 2), true, true), - (0.02::DECIMAL(38, 2), false, true), - (0.30::DECIMAL(38, 2), true, false), - (NULL::DECIMAL(38, 2), true, true) - ) input(value, active, sampled) -); ----- -90071992547410.23 90071992547409.95 DECIMAL(38,2) DECIMAL(38,2) - -# Every exact SUM input accepted by filterless binds to an exact HUGEINT -# partial; DECIMAL is bound separately above to retain its scale. +# Every integer SUM input accepted by filterless binds to an exact HUGEINT +# partial. End-to-end tests below cover values, NULLs, and DECIMAL scale. query TTTTT SELECT typeof(b.answer), typeof(t.answer), typeof(s.answer), typeof(i.answer), typeof(h.answer) FROM ( - SELECT priv_filterless_exact_sum_pair(b, true, true) AS b, - priv_filterless_exact_sum_pair(t, true, true) AS t, - priv_filterless_exact_sum_pair(s, true, true) AS s, - priv_filterless_exact_sum_pair(i, true, true) AS i, - priv_filterless_exact_sum_pair(h, true, true) AS h + SELECT priv_filterless_sum_pair(b, true, true) AS b, + priv_filterless_sum_pair(t, true, true) AS t, + priv_filterless_sum_pair(s, true, true) AS s, + priv_filterless_sum_pair(i, true, true) AS i, + priv_filterless_sum_pair(h, true, true) AS h FROM (VALUES (true::BOOLEAN, 1::TINYINT, 1::SMALLINT, 1::INTEGER, 1::HUGEINT) ) input(b, t, s, i, h) @@ -586,7 +528,7 @@ EXPLAIN SELECT SUM(amount), COUNT(*) FROM filterless_sampled_events WHERE qualifies; ---- -physical_plan :[\s\S]*priv_filterless_approx_sum_[\s\S]*pair[\s\S]*priv_filterless_count_pair[\s\S]* +physical_plan :[\s\S]*priv_filterless_sum_pair[\s\S]*priv_filterless_count_pair[\s\S]* # COUNT(expr) fuses through an explicit non-NULL marker, while COUNT(*) still # counts active rows regardless of the expression's NULL value. @@ -654,7 +596,7 @@ query II EXPLAIN SELECT SUM(integer_amount), SUM(decimal_amount), COUNT(*) FROM filterless_exact_numeric; ---- -physical_plan :[\s\S]*priv_filterless_exact_sum_p[\s\S]*air[\s\S]*priv_filterless_count_pair[\s\S]* +physical_plan :[\s\S]*priv_filterless_sum_pair[\s\S]*priv_filterless_count_pair[\s\S]* query IRTT SELECT SUM(integer_amount), SUM(decimal_amount), From 0b122ba460c6f1d0476a8e4fa08bc08f160374e0 Mon Sep 17 00:00:00 2001 From: ila Date: Tue, 1 Sep 2026 13:16:12 +0200 Subject: [PATCH 5/6] Follow aggregate type alias style --- src/aggregates/filterless_aggregate.cpp | 68 ++++++++++++------------- 1 file changed, 34 insertions(+), 34 deletions(-) diff --git a/src/aggregates/filterless_aggregate.cpp b/src/aggregates/filterless_aggregate.cpp index 220eee1..2f2a4de 100644 --- a/src/aggregates/filterless_aggregate.cpp +++ b/src/aggregates/filterless_aggregate.cpp @@ -928,15 +928,15 @@ struct FilterlessApproxSumState { // Use the same scaled-magnitude representation as PAC's approximate SUM so // cancellation cannot erase small per-PU contributions before clipping. struct FilterlessApproxSumPairOperation { - using INPUT_TYPE = double; - using STATE = FilterlessApproxSumState; - using RESULT_TYPE = double; + using input_t = double; + using state_t = FilterlessApproxSumState; + using result_t = double; static hugeint_t ApproximateScaledValue(double value) { return Hugeint::Convert(ClipApproximateMagnitude64(AsScaledMagnitude(value))); } - static void Add(STATE &state, INPUT_TYPE value) { + static void Add(state_t &state, input_t value) { if (!std::isfinite(value)) { throw InvalidInputException("filterless: per-PU SUM contribution must be finite"); } @@ -949,17 +949,17 @@ struct FilterlessApproxSumPairOperation { } } - static void Combine(const STATE &source, STATE &target) { + static void Combine(const state_t &source, state_t &target) { target.isset = target.isset || source.isset; target.positive = Hugeint::Add(target.positive, source.positive); target.negative = Hugeint::Add(target.negative, source.negative); } - static bool IsSet(const STATE &state) { + static bool IsSet(const state_t &state) { return state.isset; } - static RESULT_TYPE Finalize(const STATE &state) { + static result_t Finalize(const state_t &state) { auto scaled = Hugeint::Subtract(state.positive, state.negative); return Hugeint::Cast(scaled) / CLIP_DOUBLE_SCALE; } @@ -977,48 +977,48 @@ struct FilterlessLowerPairState { }; struct FilterlessCountPairOperation { - using INPUT_TYPE = bool; - using STATE = uint64_t; - using RESULT_TYPE = int64_t; + using input_t = bool; + using state_t = uint64_t; + using result_t = int64_t; - static void Add(STATE &state, INPUT_TYPE value) { + static void Add(state_t &state, input_t value) { state += static_cast(value); } - static void Combine(const STATE &source, STATE &target) { + static void Combine(const state_t &source, state_t &target) { target += source; } - static bool IsSet(const STATE &) { + static bool IsSet(const state_t &) { return true; } - static RESULT_TYPE Finalize(const STATE &state) { - return static_cast(state); + static result_t Finalize(const state_t &state) { + return static_cast(state); } }; template struct FilterlessExactSumPairOperation { - using INPUT_TYPE = INPUT; - using STATE = FilterlessExactSumPartialState; - using RESULT_TYPE = hugeint_t; + using input_t = INPUT; + using state_t = FilterlessExactSumPartialState; + using result_t = hugeint_t; - static void Add(STATE &state, INPUT_TYPE value) { + static void Add(state_t &state, input_t value) { state.isset = true; state.value = Hugeint::Add(state.value, ToHugeint(value)); } - static void Combine(const STATE &source, STATE &target) { + static void Combine(const state_t &source, state_t &target) { target.isset = target.isset || source.isset; target.value = Hugeint::Add(target.value, source.value); } - static bool IsSet(const STATE &state) { + static bool IsSet(const state_t &state) { return state.isset; } - static RESULT_TYPE Finalize(const STATE &state) { + static result_t Finalize(const state_t &state) { return state.value; } }; @@ -1029,12 +1029,12 @@ static LogicalType FilterlessLowerPairType(const LogicalType &value_type) { template static idx_t FilterlessLowerPairStateSize(const AggregateFunction &) { - return sizeof(FilterlessLowerPairState); + return sizeof(FilterlessLowerPairState); } template static void FilterlessLowerPairInitialize(const AggregateFunction &, data_ptr_t state_p) { - memset(state_p, 0, sizeof(FilterlessLowerPairState)); + memset(state_p, 0, sizeof(FilterlessLowerPairState)); } template @@ -1043,7 +1043,7 @@ static void FilterlessLowerPairUpdateRows(Vector inputs[], idx_t count, STATE_GE inputs[0].ToUnifiedFormat(count, value_data); inputs[1].ToUnifiedFormat(count, active_data); inputs[2].ToUnifiedFormat(count, sampled_data); - auto values = UnifiedVectorFormat::GetData(value_data); + auto values = UnifiedVectorFormat::GetData(value_data); auto active = UnifiedVectorFormat::GetData(active_data); auto sampled = UnifiedVectorFormat::GetData(sampled_data); for (idx_t row = 0; row < count; row++) { @@ -1069,7 +1069,7 @@ static void FilterlessLowerPairUpdateRows(Vector inputs[], idx_t count, STATE_GE template static void FilterlessLowerPairUpdate(Vector inputs[], AggregateInputData &, idx_t, data_ptr_t state_p, idx_t count) { - auto state = reinterpret_cast *>(state_p); + auto state = reinterpret_cast *>(state_p); FilterlessLowerPairUpdateRows(inputs, count, [state](idx_t) { return state; }); } @@ -1078,16 +1078,16 @@ static void FilterlessLowerPairScatterUpdate(Vector inputs[], AggregateInputData idx_t count) { UnifiedVectorFormat state_data; states.ToUnifiedFormat(count, state_data); - auto state_ptrs = UnifiedVectorFormat::GetData *>(state_data); + auto state_ptrs = UnifiedVectorFormat::GetData *>(state_data); FilterlessLowerPairUpdateRows(inputs, count, [&](idx_t row) { return state_ptrs[state_data.sel->get_index(row)]; }); } template static void FilterlessLowerPairCombine(Vector &source, Vector &target, AggregateInputData &, idx_t count) { - using PAIR_STATE = FilterlessLowerPairState; - auto sources = FlatVector::GetData(source); - auto targets = FlatVector::GetData(target); + using pair_state_t = FilterlessLowerPairState; + auto sources = FlatVector::GetData(source); + auto targets = FlatVector::GetData(target); for (idx_t i = 0; i < count; i++) { OPERATION::Combine(sources[i]->answer, targets[i]->answer); OPERATION::Combine(sources[i]->histogram, targets[i]->histogram); @@ -1097,11 +1097,11 @@ static void FilterlessLowerPairCombine(Vector &source, Vector &target, Aggregate template static void FilterlessLowerPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, idx_t offset) { - using PAIR_STATE = FilterlessLowerPairState; - auto state_ptrs = FlatVector::GetData(states); + using pair_state_t = FilterlessLowerPairState; + auto state_ptrs = FlatVector::GetData(states); auto &children = StructVector::GetEntries(result); - auto answers = FlatVector::GetData(*children[0]); - auto histograms = FlatVector::GetData(*children[1]); + auto answers = FlatVector::GetData(*children[0]); + auto histograms = FlatVector::GetData(*children[1]); for (idx_t i = 0; i < count; i++) { auto row = offset + i; if (OPERATION::IsSet(state_ptrs[i]->answer)) { From 12f35cd39b8a1953296ff365ca8c82b020b38969 Mon Sep 17 00:00:00 2001 From: ila Date: Tue, 1 Sep 2026 22:07:03 +0200 Subject: [PATCH 6/6] Fuse filterless AVG release --- docs/dp/filterless_encoded_pu.md | 5 +- src/aggregates/filterless_aggregate.cpp | 444 +++++++++++++++++------- src/compiler/privacy_mechanisms.cpp | 87 ++--- test/sql/dp_filterless.test | 47 ++- 4 files changed, 383 insertions(+), 200 deletions(-) diff --git a/docs/dp/filterless_encoded_pu.md b/docs/dp/filterless_encoded_pu.md index 366a32c..46eef23 100644 --- a/docs/dp/filterless_encoded_pu.md +++ b/docs/dp/filterless_encoded_pu.md @@ -83,9 +83,8 @@ FROM per_pu_partials; `filterless_sum_debug`, `filterless_count_debug`, and `filterless_avg_debug` return selected bounds, clipped values, noise scales, bin indices, and active/sample counts. The compiler uses overloads -that also receive the component epsilon, `C_u`, and a stable group nonce. The nonce includes the -normalized SQL hash, so different predicates do not accidentally reuse cancelable value, histogram, -or partition-selection noise; an identical query remains deterministic for a fixed privacy seed. +that also receive the component epsilon and `C_u`. Every release uses fresh secure randomness, so +the aggregate does not carry a query or group nonce. ## Settings diff --git a/src/aggregates/filterless_aggregate.cpp b/src/aggregates/filterless_aggregate.cpp index 2f2a4de..b3743e3 100644 --- a/src/aggregates/filterless_aggregate.cpp +++ b/src/aggregates/filterless_aggregate.cpp @@ -90,15 +90,11 @@ struct FilterlessComponentState { struct FilterlessState { FilterlessComponentState component; - uint64_t nonce; - bool nonce_set; }; struct FilterlessAvgState { FilterlessComponentState sum_component; FilterlessComponentState count_component; - uint64_t nonce; - bool nonce_set; }; // With the shared 2^-27 anchor and factor-4 levels, 80 bins reach 2^133 and therefore cover @@ -125,8 +121,6 @@ struct FilterlessExactComponentState { struct FilterlessExactState { FilterlessExactComponentState component; - uint64_t nonce; - bool nonce_set; }; struct FilterlessBindData : public FunctionData { @@ -137,7 +131,6 @@ struct FilterlessBindData : public FunctionData { double epsilon; double bounds_fraction; double max_groups; - bool has_explicit_config; double input_scale; bool approximate_values; idx_t exact_bin_count; @@ -153,9 +146,9 @@ struct FilterlessBindData : public FunctionData { return other && sample_bits == other->sample_bits && clip_support == other->clip_support && noise_enabled == other->noise_enabled && epsilon == other->epsilon && bounds_fraction == other->bounds_fraction && max_groups == other->max_groups && - has_explicit_config == other->has_explicit_config && input_scale == other->input_scale && - approximate_values == other->approximate_values && exact_bin_count == other->exact_bin_count && - exact_output_min == other->exact_output_min && exact_output_max == other->exact_output_max; + input_scale == other->input_scale && approximate_values == other->approximate_values && + exact_bin_count == other->exact_bin_count && exact_output_min == other->exact_output_min && + exact_output_max == other->exact_output_max; } }; @@ -218,7 +211,6 @@ static unique_ptr BindFilterless(ClientContext &context, vectorepsilon = epsilon; result->bounds_fraction = settings.bounds_epsilon_fraction; result->max_groups = max_groups; - result->has_explicit_config = has_explicit_config; result->input_scale = 1.0; result->approximate_values = approximate_values; result->exact_bin_count = exact_bin_count; @@ -289,7 +281,7 @@ static FilterlessBin &GetBin(FilterlessComponentState &state, double value, uint return negative ? EnsureBins(state.negative, allocator)[index] : EnsureBins(state.positive, allocator)[index]; } -static void UpdateComponent(FilterlessComponentState &state, uint64_t pu_hash, bool active, bool answer_valid, +static void UpdateComponent(FilterlessComponentState &state, bool active, bool sampled, bool answer_valid, double answer_value, bool histogram_valid, double histogram_value, const FilterlessBindData &bind, ArenaAllocator &allocator, bool approximate_values) { if (active) { @@ -307,7 +299,7 @@ static void UpdateComponent(FilterlessComponentState &state, uint64_t pu_hash, b answer_bin.answer_count++; } } - if (FilterlessPuIsSampled(pu_hash, bind.sample_bits) && histogram_valid) { + if (sampled && histogram_valid) { if (!std::isfinite(histogram_value)) { throw InvalidInputException("filterless: histogram aggregate contribution must be finite"); } @@ -338,7 +330,7 @@ static void CombineComponent(const FilterlessComponentState &source, FilterlessC } template -static int FindSupportedBin(const BIN_TYPE *bins, idx_t bin_count, const FilterlessBindData &bind, uint64_t, +static int FindSupportedBin(const BIN_TYPE *bins, idx_t bin_count, const FilterlessBindData &bind, double histogram_epsilon, double &selected_support) { D_ASSERT(bin_count <= FILTERLESS_EXACT_BIN_COUNT); double scale = @@ -408,17 +400,15 @@ static double ClipComponent(const FilterlessComponentState &state, int negative_ } static FilterlessResult FinalizeComponent(const FilterlessComponentState &state, const FilterlessBindData &bind, - uint64_t nonce_base, double epsilon, bool nonnegative) { + double epsilon, bool nonnegative) { double histogram_epsilon = epsilon * bind.bounds_fraction; double value_epsilon = epsilon * (1.0 - bind.bounds_fraction); double negative_support = 0.0; double positive_support = 0.0; - int positive_bin = - FindSupportedBin(state.positive, CLIP_NUM_LEVELS_64, bind, nonce_base, histogram_epsilon, positive_support); - int negative_bin = nonnegative - ? -1 - : FindSupportedBin(state.negative, CLIP_NUM_LEVELS_64, bind, nonce_base + CLIP_NUM_LEVELS_64, - histogram_epsilon, negative_support); + int positive_bin = FindSupportedBin(state.positive, CLIP_NUM_LEVELS_64, bind, histogram_epsilon, positive_support); + int negative_bin = + nonnegative ? -1 + : FindSupportedBin(state.negative, CLIP_NUM_LEVELS_64, bind, histogram_epsilon, negative_support); double positive_bound = BinUpperBound(positive_bin); double negative_bound = BinUpperBound(negative_bin); double clipped = ClipComponent(state, negative_bin, positive_bin); @@ -436,8 +426,8 @@ static FilterlessResult FinalizeComponent(const FilterlessComponentState &state, state.sampled_contributions}; } -static idx_t ExactBinIndex(hugeint_t value, const FilterlessBindData &bind) { - double magnitude = std::abs(Hugeint::Cast(value)) / bind.input_scale; +static idx_t ExactBinIndex(hugeint_t value, double input_scale) { + double magnitude = std::abs(Hugeint::Cast(value)) / input_scale; if (!std::isfinite(magnitude)) { throw InvalidInputException("filterless: exact aggregate contribution is outside the supported numeric range"); } @@ -470,10 +460,10 @@ static FilterlessExactBin *EnsureExactBins(FilterlessExactBin *&bins, ArenaAlloc return bins; } -static FilterlessExactBin &GetExactBin(FilterlessExactComponentState &state, hugeint_t value, - const FilterlessBindData &bind, ArenaAllocator &allocator) { +static FilterlessExactBin &GetExactBin(FilterlessExactComponentState &state, hugeint_t value, double input_scale, + ArenaAllocator &allocator) { bool negative = value < 0; - auto index = ExactBinIndex(value, bind); + auto index = ExactBinIndex(value, input_scale); return negative ? EnsureExactBins(state.negative, allocator)[index] : EnsureExactBins(state.positive, allocator)[index]; } @@ -492,21 +482,21 @@ static hugeint_t ToHugeint(INPUT_TYPE value) { } template -static void UpdateExactComponent(FilterlessExactComponentState &state, uint64_t pu_hash, bool active, bool answer_valid, +static void UpdateExactComponent(FilterlessExactComponentState &state, bool active, bool sampled, bool answer_valid, INPUT_TYPE answer_value, bool histogram_valid, INPUT_TYPE histogram_value, - const FilterlessBindData &bind, ArenaAllocator &allocator) { + const FilterlessBindData &bind, ArenaAllocator &allocator, double input_scale) { if (active) { state.active_contributions++; if (answer_valid) { auto exact_answer = ToHugeint(answer_value); - auto &answer_bin = GetExactBin(state, exact_answer, bind, allocator); + auto &answer_bin = GetExactBin(state, exact_answer, input_scale, allocator); answer_bin.answer_sum = Hugeint::Add(answer_bin.answer_sum, exact_answer); answer_bin.answer_count++; } } - if (FilterlessPuIsSampled(pu_hash, bind.sample_bits) && histogram_valid) { + if (sampled && histogram_valid) { auto exact_histogram = ToHugeint(histogram_value); - GetExactBin(state, exact_histogram, bind, allocator).support += bind.sample_weight; + GetExactBin(state, exact_histogram, input_scale, allocator).support += bind.sample_weight; state.sampled_contributions++; } } @@ -531,11 +521,11 @@ static void CombineExactComponent(const FilterlessExactComponentState &source, F target.sampled_contributions += source.sampled_contributions; } -static hugeint_t ExactClippingBound(int bin, const FilterlessBindData &bind) { +static hugeint_t ExactClippingBound(int bin, double input_scale) { if (bin < 0) { return hugeint_t(0); } - double scaled_bound = std::ceil(ExactBinUpperBound(bin) * bind.input_scale); + double scaled_bound = std::ceil(ExactBinUpperBound(bin) * input_scale); hugeint_t result; if (!std::isfinite(scaled_bound) || !Hugeint::TryConvert(scaled_bound, result)) { return NumericLimits::Maximum(); @@ -544,9 +534,9 @@ static hugeint_t ExactClippingBound(int bin, const FilterlessBindData &bind) { } static hugeint_t ClipExactComponent(const FilterlessExactComponentState &state, int negative_bin, int positive_bin, - const FilterlessBindData &bind) { - auto positive_bound = ExactClippingBound(positive_bin, bind); - auto negative_bound = ExactClippingBound(negative_bin, bind); + double input_scale) { + auto positive_bound = ExactClippingBound(positive_bin, input_scale); + auto negative_bound = ExactClippingBound(negative_bin, input_scale); hugeint_t result(0); for (int i = 0; i < FILTERLESS_EXACT_BIN_COUNT; i++) { if (state.positive) { @@ -568,19 +558,16 @@ struct FilterlessExactResult { }; static FilterlessExactResult FinalizeExactComponent(const FilterlessExactComponentState &state, - const FilterlessBindData &bind, uint64_t nonce_base, double epsilon, - bool nonnegative) { + const FilterlessBindData &bind, double epsilon, bool nonnegative, + idx_t bin_count, double input_scale) { double histogram_epsilon = epsilon * bind.bounds_fraction; double value_epsilon = epsilon * (1.0 - bind.bounds_fraction); double ignored_support; - idx_t bin_count = bind.exact_bin_count; - int positive_bin = - FindSupportedBin(state.positive, bin_count, bind, nonce_base, histogram_epsilon, ignored_support); - int negative_bin = nonnegative ? -1 - : FindSupportedBin(state.negative, bin_count, bind, nonce_base + bin_count, - histogram_epsilon, ignored_support); + int positive_bin = FindSupportedBin(state.positive, bin_count, bind, histogram_epsilon, ignored_support); + int negative_bin = + nonnegative ? -1 : FindSupportedBin(state.negative, bin_count, bind, histogram_epsilon, ignored_support); double bound = std::max(ExactBinUpperBound(negative_bin), ExactBinUpperBound(positive_bin)); - return {ClipExactComponent(state, negative_bin, positive_bin, bind), + return {ClipExactComponent(state, negative_bin, positive_bin, input_scale), SaturatingNoiseScale(static_cast(bound) * bind.max_groups, value_epsilon)}; } @@ -608,26 +595,16 @@ static void FilterlessExactInitialize(const AggregateFunction &, data_ptr_t stat memset(state_p, 0, sizeof(FilterlessExactState)); } -static void SetFilterlessNonce(uint64_t value, uint64_t &nonce, bool &nonce_set) { - if (nonce_set && nonce != value) { - throw InvalidInputException("filterless: noise nonce must be constant within each aggregate group"); - } - nonce = value; - nonce_set = true; -} - template struct FilterlessInputVectors { UnifiedVectorFormat pu; UnifiedVectorFormat active; UnifiedVectorFormat values[VALUE_COUNT]; - UnifiedVectorFormat nonce; const uint64_t *pu_values; const bool *active_values; const INPUT_TYPE *numeric_values[VALUE_COUNT]; - const uint64_t *nonce_values; - FilterlessInputVectors(Vector inputs[], idx_t count, bool has_explicit_config) : nonce_values(nullptr) { + FilterlessInputVectors(Vector inputs[], idx_t count) { inputs[0].ToUnifiedFormat(count, pu); inputs[1].ToUnifiedFormat(count, active); pu_values = UnifiedVectorFormat::GetData(pu); @@ -636,19 +613,12 @@ struct FilterlessInputVectors { inputs[2 + i].ToUnifiedFormat(count, values[i]); numeric_values[i] = UnifiedVectorFormat::GetData(values[i]); } - if (has_explicit_config) { - inputs[VALUE_COUNT + 4].ToUnifiedFormat(count, nonce); - nonce_values = UnifiedVectorFormat::GetData(nonce); - } } - bool RequiredValuesAreValid(idx_t row, bool has_explicit_config) const { + bool RequiredValuesAreValid(idx_t row) const { auto pu_index = pu.sel->get_index(row); auto active_index = active.sel->get_index(row); - if (!pu.validity.RowIsValid(pu_index) || !active.validity.RowIsValid(active_index)) { - return false; - } - return !has_explicit_config || nonce.validity.RowIsValid(nonce.sel->get_index(row)); + return pu.validity.RowIsValid(pu_index) && active.validity.RowIsValid(active_index); } bool ValueIsValid(idx_t value_index, idx_t row) const { @@ -668,21 +638,23 @@ static void UpdateFilterlessStateRow(FilterlessState &state, const FilterlessInp const FilterlessBindData &bind, ArenaAllocator &allocator) { auto pu_index = input.pu.sel->get_index(row); auto active_index = input.active.sel->get_index(row); - UpdateComponent(state.component, input.pu_values[pu_index], input.active_values[active_index], - input.ValueIsValid(0, row), input.ValueOrZero(0, row), input.ValueIsValid(1, row), - input.ValueOrZero(1, row), bind, allocator, bind.approximate_values); + bool sampled = FilterlessPuIsSampled(input.pu_values[pu_index], bind.sample_bits); + UpdateComponent(state.component, input.active_values[active_index], sampled, input.ValueIsValid(0, row), + input.ValueOrZero(0, row), input.ValueIsValid(1, row), input.ValueOrZero(1, row), bind, allocator, + bind.approximate_values); } static void UpdateFilterlessStateRow(FilterlessAvgState &state, const FilterlessInputVectors<4, double> &input, idx_t row, const FilterlessBindData &bind, ArenaAllocator &allocator) { auto pu_index = input.pu.sel->get_index(row); auto active_index = input.active.sel->get_index(row); - UpdateComponent(state.sum_component, input.pu_values[pu_index], input.active_values[active_index], - input.ValueIsValid(0, row), input.ValueOrZero(0, row), input.ValueIsValid(2, row), - input.ValueOrZero(2, row), bind, allocator, true); - UpdateComponent(state.count_component, input.pu_values[pu_index], input.active_values[active_index], - input.ValueIsValid(1, row), input.ValueOrZero(1, row), input.ValueIsValid(3, row), - input.ValueOrZero(3, row), bind, allocator, false); + bool sampled = FilterlessPuIsSampled(input.pu_values[pu_index], bind.sample_bits); + UpdateComponent(state.sum_component, input.active_values[active_index], sampled, input.ValueIsValid(0, row), + input.ValueOrZero(0, row), input.ValueIsValid(2, row), input.ValueOrZero(2, row), bind, allocator, + true); + UpdateComponent(state.count_component, input.active_values[active_index], sampled, input.ValueIsValid(1, row), + input.ValueOrZero(1, row), input.ValueIsValid(3, row), input.ValueOrZero(3, row), bind, allocator, + false); } template @@ -690,24 +662,21 @@ static void UpdateFilterlessStateRow(FilterlessExactState &state, const Filterle idx_t row, const FilterlessBindData &bind, ArenaAllocator &allocator) { auto pu_index = input.pu.sel->get_index(row); auto active_index = input.active.sel->get_index(row); - UpdateExactComponent(state.component, input.pu_values[pu_index], input.active_values[active_index], - input.ValueIsValid(0, row), input.ValueOrZero(0, row), input.ValueIsValid(1, row), - input.ValueOrZero(1, row), bind, allocator); + bool sampled = FilterlessPuIsSampled(input.pu_values[pu_index], bind.sample_bits); + UpdateExactComponent(state.component, input.active_values[active_index], sampled, input.ValueIsValid(0, row), + input.ValueOrZero(0, row), input.ValueIsValid(1, row), input.ValueOrZero(1, row), bind, + allocator, bind.input_scale); } template static void FilterlessUpdateRows(Vector inputs[], AggregateInputData &aggr, idx_t count, STATE_GETTER get_state) { auto &bind = aggr.bind_data->Cast(); - FilterlessInputVectors input(inputs, count, bind.has_explicit_config); + FilterlessInputVectors input(inputs, count); for (idx_t row = 0; row < count; row++) { - if (!input.RequiredValuesAreValid(row, bind.has_explicit_config)) { + if (!input.RequiredValuesAreValid(row)) { continue; } auto state = get_state(row); - if (bind.has_explicit_config) { - auto nonce_index = input.nonce.sel->get_index(row); - SetFilterlessNonce(input.nonce_values[nonce_index], state->nonce, state->nonce_set); - } UpdateFilterlessStateRow(*state, input, row, bind, aggr.allocator); } } @@ -759,9 +728,6 @@ static void FilterlessCombine(Vector &source, Vector &target, AggregateInputData auto targets = FlatVector::GetData(target); for (idx_t i = 0; i < count; i++) { CombineComponent(sources[i]->component, targets[i]->component, input.allocator); - if (sources[i]->nonce_set) { - SetFilterlessNonce(sources[i]->nonce, targets[i]->nonce, targets[i]->nonce_set); - } } } @@ -771,9 +737,6 @@ static void FilterlessAvgCombine(Vector &source, Vector &target, AggregateInputD for (idx_t i = 0; i < count; i++) { CombineComponent(sources[i]->sum_component, targets[i]->sum_component, input.allocator); CombineComponent(sources[i]->count_component, targets[i]->count_component, input.allocator); - if (sources[i]->nonce_set) { - SetFilterlessNonce(sources[i]->nonce, targets[i]->nonce, targets[i]->nonce_set); - } } } @@ -782,9 +745,6 @@ static void FilterlessExactCombine(Vector &source, Vector &target, AggregateInpu auto targets = FlatVector::GetData(target); for (idx_t i = 0; i < count; i++) { CombineExactComponent(sources[i]->component, targets[i]->component, input.allocator); - if (sources[i]->nonce_set) { - SetFilterlessNonce(sources[i]->nonce, targets[i]->nonce, targets[i]->nonce_set); - } } } @@ -826,8 +786,7 @@ static void FilterlessFinalize(Vector &states, AggregateInputData &input, Vector auto &bind = input.bind_data->Cast(); auto result_data = DEBUG ? nullptr : FlatVector::GetData(result); for (idx_t i = 0; i < count; i++) { - uint64_t nonce = state_ptrs[i]->nonce_set ? state_ptrs[i]->nonce : 0; - auto value = FinalizeComponent(state_ptrs[i]->component, bind, nonce * 1024, bind.epsilon, COUNT); + auto value = FinalizeComponent(state_ptrs[i]->component, bind, bind.epsilon, COUNT); if (DEBUG) { WriteDebugResult(result, offset + i, value); } else { @@ -844,12 +803,9 @@ static void FilterlessAvgFinalize(Vector &states, AggregateInputData &input, Vec auto &bind = input.bind_data->Cast(); auto result_data = DEBUG ? nullptr : FlatVector::GetData(result); for (idx_t i = 0; i < count; i++) { - uint64_t nonce = state_ptrs[i]->nonce_set ? state_ptrs[i]->nonce : 0; double component_epsilon = bind.epsilon / 2.0; - auto sum = FinalizeComponent(state_ptrs[i]->sum_component, bind, nonce * 2048, component_epsilon, false); - auto denominator = FinalizeComponent(state_ptrs[i]->count_component, bind, - nonce * 2048 + static_cast(2 * CLIP_NUM_LEVELS_64 + 1), - component_epsilon, true); + auto sum = FinalizeComponent(state_ptrs[i]->sum_component, bind, component_epsilon, false); + auto denominator = FinalizeComponent(state_ptrs[i]->count_component, bind, component_epsilon, true); double noised_sum = bind.noise_enabled ? AddDpLaplaceNoise(sum.clipped_value, sum.noise_scale) : sum.clipped_value; double noised_count = bind.noise_enabled ? AddDpLaplaceNoise(denominator.clipped_value, denominator.noise_scale) @@ -900,6 +856,18 @@ static hugeint_t ClampExactResult(hugeint_t value, const FilterlessBindData &bin return std::max(bind.exact_output_min, std::min(value, bind.exact_output_max)); } +static hugeint_t ReleaseExactComponent(const FilterlessExactComponentState &state, const FilterlessBindData &bind, + double epsilon, bool nonnegative, idx_t bin_count) { + auto value = FinalizeExactComponent(state, bind, epsilon, nonnegative, bin_count, bind.input_scale); + auto released = value.clipped_value; + if (bind.noise_enabled) { + double noise = AddDpLaplaceNoise(0.0, value.noise_scale); + auto scaled_noise = SaturatingHugeintFromDouble(noise * bind.input_scale); + released = SaturatingHugeintAdd(released, scaled_noise); + } + return ClampExactResult(released, bind); +} + template static void FilterlessExactFinalize(Vector &states, AggregateInputData &input, Vector &result, idx_t count, idx_t offset) { @@ -907,15 +875,169 @@ static void FilterlessExactFinalize(Vector &states, AggregateInputData &input, V auto result_data = FlatVector::GetData(result); auto &bind = input.bind_data->Cast(); for (idx_t i = 0; i < count; i++) { - uint64_t nonce = state_ptrs[i]->nonce_set ? state_ptrs[i]->nonce : 0; - auto value = FinalizeExactComponent(state_ptrs[i]->component, bind, nonce * 1024, bind.epsilon, COUNT); - auto released = value.clipped_value; + auto released = + ReleaseExactComponent(state_ptrs[i]->component, bind, bind.epsilon, COUNT, bind.exact_bin_count); + result_data[offset + i] = CastExactResult(released); + } +} + +struct FilterlessApproxAvgSumOperation { + using input_t = double; + using component_t = FilterlessComponentState; + + static void Update(component_t &state, bool active, bool sampled, bool answer_valid, input_t answer, + bool histogram_valid, input_t histogram, const FilterlessBindData &bind, + ArenaAllocator &allocator) { + UpdateComponent(state, active, sampled, answer_valid, answer, histogram_valid, histogram, bind, allocator, + true); + } + + static void Combine(const component_t &source, component_t &target, ArenaAllocator &allocator) { + CombineComponent(source, target, allocator); + } + + static double Release(const component_t &state, const FilterlessBindData &bind, double epsilon) { + auto value = FinalizeComponent(state, bind, epsilon, false); + return bind.noise_enabled ? AddDpLaplaceNoise(value.clipped_value, value.noise_scale) : value.clipped_value; + } +}; + +template +struct FilterlessExactAvgSumOperation { + using input_t = INPUT_TYPE; + using component_t = FilterlessExactComponentState; + + static void Update(component_t &state, bool active, bool sampled, bool answer_valid, input_t answer, + bool histogram_valid, input_t histogram, const FilterlessBindData &bind, + ArenaAllocator &allocator) { + UpdateExactComponent(state, active, sampled, answer_valid, answer, histogram_valid, histogram, bind, allocator, + bind.input_scale); + } + + static void Combine(const component_t &source, component_t &target, ArenaAllocator &allocator) { + CombineExactComponent(source, target, allocator); + } + + static double Release(const component_t &state, const FilterlessBindData &bind, double epsilon) { + auto released = ReleaseExactComponent(state, bind, epsilon, false, bind.exact_bin_count); + return Hugeint::Cast(released) / bind.input_scale; + } +}; + +// The compiler's upper AVG consumes the already paired per-PU SUM and COUNT partials in one pass. +// COUNT stays exact so fusion preserves the previous SUM / BIGINT COUNT release semantics. +template +struct FilterlessFusedAvgState { + typename SUM_OPERATION::component_t sum_component; + FilterlessExactComponentState count_component; +}; + +template +static idx_t FilterlessFusedAvgStateSize(const AggregateFunction &) { + return sizeof(FilterlessFusedAvgState); +} + +template +static void FilterlessFusedAvgInitialize(const AggregateFunction &, data_ptr_t state_p) { + memset(state_p, 0, sizeof(FilterlessFusedAvgState)); +} + +template +static void FilterlessFusedAvgUpdateRows(Vector inputs[], AggregateInputData &aggr, idx_t count, + STATE_GETTER get_state) { + UnifiedVectorFormat pu_data, active_data, answer_sum_data, answer_count_data, histogram_sum_data, + histogram_count_data; + inputs[0].ToUnifiedFormat(count, pu_data); + inputs[1].ToUnifiedFormat(count, active_data); + inputs[2].ToUnifiedFormat(count, answer_sum_data); + inputs[3].ToUnifiedFormat(count, answer_count_data); + inputs[4].ToUnifiedFormat(count, histogram_sum_data); + inputs[5].ToUnifiedFormat(count, histogram_count_data); + auto pus = UnifiedVectorFormat::GetData(pu_data); + auto active = UnifiedVectorFormat::GetData(active_data); + auto answer_sums = UnifiedVectorFormat::GetData(answer_sum_data); + auto answer_counts = UnifiedVectorFormat::GetData(answer_count_data); + auto histogram_sums = UnifiedVectorFormat::GetData(histogram_sum_data); + auto histogram_counts = UnifiedVectorFormat::GetData(histogram_count_data); + auto &bind = aggr.bind_data->Cast(); + for (idx_t row = 0; row < count; row++) { + auto pu_index = pu_data.sel->get_index(row); + auto active_index = active_data.sel->get_index(row); + if (!pu_data.validity.RowIsValid(pu_index) || !active_data.validity.RowIsValid(active_index)) { + continue; + } + auto answer_sum_index = answer_sum_data.sel->get_index(row); + auto answer_count_index = answer_count_data.sel->get_index(row); + auto histogram_sum_index = histogram_sum_data.sel->get_index(row); + auto histogram_count_index = histogram_count_data.sel->get_index(row); + bool answer_sum_valid = answer_sum_data.validity.RowIsValid(answer_sum_index); + bool answer_count_valid = answer_count_data.validity.RowIsValid(answer_count_index); + bool histogram_sum_valid = histogram_sum_data.validity.RowIsValid(histogram_sum_index); + bool histogram_count_valid = histogram_count_data.validity.RowIsValid(histogram_count_index); + bool sampled = FilterlessPuIsSampled(pus[pu_index], bind.sample_bits); + auto state = get_state(row); + SUM_OPERATION::Update( + state->sum_component, active[active_index], sampled, answer_sum_valid, + answer_sum_valid ? answer_sums[answer_sum_index] : typename SUM_OPERATION::input_t(0), histogram_sum_valid, + histogram_sum_valid ? histogram_sums[histogram_sum_index] : typename SUM_OPERATION::input_t(0), bind, + aggr.allocator); + UpdateExactComponent(state->count_component, active[active_index], sampled, answer_count_valid, + answer_count_valid ? answer_counts[answer_count_index] : 0, histogram_count_valid, + histogram_count_valid ? histogram_counts[histogram_count_index] : 0, bind, aggr.allocator, + 1.0); + } +} + +template +static void FilterlessFusedAvgUpdate(Vector inputs[], AggregateInputData &aggr, idx_t, data_ptr_t state_p, + idx_t count) { + auto state = reinterpret_cast *>(state_p); + FilterlessFusedAvgUpdateRows(inputs, aggr, count, [state](idx_t) { return state; }); +} + +template +static void FilterlessFusedAvgScatterUpdate(Vector inputs[], AggregateInputData &aggr, idx_t, Vector &states, + idx_t count) { + UnifiedVectorFormat state_data; + states.ToUnifiedFormat(count, state_data); + auto state_ptrs = UnifiedVectorFormat::GetData *>(state_data); + FilterlessFusedAvgUpdateRows(inputs, aggr, count, + [&](idx_t row) { return state_ptrs[state_data.sel->get_index(row)]; }); +} + +template +static void FilterlessFusedAvgCombine(Vector &source, Vector &target, AggregateInputData &input, idx_t count) { + auto sources = FlatVector::GetData *>(source); + auto targets = FlatVector::GetData *>(target); + for (idx_t i = 0; i < count; i++) { + SUM_OPERATION::Combine(sources[i]->sum_component, targets[i]->sum_component, input.allocator); + CombineExactComponent(sources[i]->count_component, targets[i]->count_component, input.allocator); + } +} + +template +static void FilterlessFusedAvgFinalize(Vector &states, AggregateInputData &input, Vector &result, idx_t count, + idx_t offset) { + auto state_ptrs = FlatVector::GetData *>(states); + auto result_data = FlatVector::GetData(result); + auto &validity = FlatVector::Validity(result); + auto &bind = input.bind_data->Cast(); + double component_epsilon = bind.epsilon / 2.0; + for (idx_t i = 0; i < count; i++) { + double sum = SUM_OPERATION::Release(state_ptrs[i]->sum_component, bind, component_epsilon); + auto count_value = FinalizeExactComponent(state_ptrs[i]->count_component, bind, component_epsilon, true, + FILTERLESS_COUNT_BIN_COUNT, 1.0); + auto released_count = count_value.clipped_value; if (bind.noise_enabled) { - double noise = AddDpLaplaceNoise(0.0, value.noise_scale); - auto scaled_noise = SaturatingHugeintFromDouble(noise * bind.input_scale); - released = SaturatingHugeintAdd(released, scaled_noise); + auto noise = SaturatingHugeintFromDouble(AddDpLaplaceNoise(0.0, count_value.noise_scale)); + released_count = SaturatingHugeintAdd(released_count, noise); + } + auto count_result = CastExactResult(released_count); + if (count_result <= 0) { + validity.SetInvalid(offset + i); + } else { + result_data[offset + i] = sum / static_cast(count_result); } - result_data[offset + i] = CastExactResult(ClampExactResult(released, bind)); } } @@ -1157,7 +1279,6 @@ static AggregateFunction MakeFilterlessExactFunction(const string &name, const L if (explicit_config) { arguments.push_back(LogicalType::DOUBLE); arguments.push_back(LogicalType::DOUBLE); - arguments.push_back(LogicalType::UBIGINT); } return AggregateFunction(name, std::move(arguments), return_type, FilterlessExactStateSize, FilterlessExactInitialize, FilterlessExactScatterUpdate, @@ -1193,9 +1314,15 @@ static AggregateFunction MakeFilterlessDecimalSumFunction(const LogicalType &inp } } +static void ConfigureFilterlessDecimalBind(FilterlessBindData &bind, const LogicalType &input_type) { + bind.input_scale = std::pow(10.0, DecimalType::GetScale(input_type)); + bind.exact_output_max = Hugeint::Subtract(Hugeint::POWERS_OF_TEN[Decimal::MAX_WIDTH_DECIMAL], hugeint_t(1)); + bind.exact_output_min = Hugeint::Negate(bind.exact_output_max); +} + static unique_ptr BindFilterlessDecimalSum(ClientContext &context, AggregateFunction &function, vector> &arguments) { - if (arguments.size() != 4 && arguments.size() != 7) { + if (arguments.size() != 4 && arguments.size() != 6) { throw InternalException("filterless_sum: unexpected DECIMAL argument count"); } auto input_type = arguments[2]->return_type; @@ -1203,12 +1330,9 @@ static unique_ptr BindFilterlessDecimalSum(ClientContext &context, throw InvalidInputException("filterless_sum: answer and histogram DECIMAL types must match"); } auto return_type = LogicalType::DECIMAL(Decimal::MAX_WIDTH_DECIMAL, DecimalType::GetScale(input_type)); - function = MakeFilterlessDecimalSumFunction(input_type, return_type, arguments.size() == 7); + function = MakeFilterlessDecimalSumFunction(input_type, return_type, arguments.size() == 6); auto result = BindFilterless(context, arguments, 4, false, FILTERLESS_HUGEINT_BIN_COUNT); - auto &bind = result->Cast(); - bind.input_scale = std::pow(10.0, DecimalType::GetScale(input_type)); - bind.exact_output_max = Hugeint::Subtract(Hugeint::POWERS_OF_TEN[Decimal::MAX_WIDTH_DECIMAL], hugeint_t(1)); - bind.exact_output_min = Hugeint::Negate(bind.exact_output_max); + ConfigureFilterlessDecimalBind(result->Cast(), input_type); return result; } @@ -1218,12 +1342,12 @@ static void AddSumCountOverloads(AggregateFunctionSet &set, const string &name, AggregateFunction(name, {LogicalType::UBIGINT, LogicalType::BOOLEAN, LogicalType::DOUBLE, LogicalType::DOUBLE}, return_type, FilterlessStateSize, FilterlessInitialize, FilterlessScatterUpdate, FilterlessCombine, finalize, FunctionNullHandling::SPECIAL_HANDLING, FilterlessUpdate, bind)); - set.AddFunction( - AggregateFunction(name, - {LogicalType::UBIGINT, LogicalType::BOOLEAN, LogicalType::DOUBLE, LogicalType::DOUBLE, - LogicalType::DOUBLE, LogicalType::DOUBLE, LogicalType::UBIGINT}, - return_type, FilterlessStateSize, FilterlessInitialize, FilterlessScatterUpdate, - FilterlessCombine, finalize, FunctionNullHandling::SPECIAL_HANDLING, FilterlessUpdate, bind)); + set.AddFunction(AggregateFunction(name, + {LogicalType::UBIGINT, LogicalType::BOOLEAN, LogicalType::DOUBLE, + LogicalType::DOUBLE, LogicalType::DOUBLE, LogicalType::DOUBLE}, + return_type, FilterlessStateSize, FilterlessInitialize, FilterlessScatterUpdate, + FilterlessCombine, finalize, FunctionNullHandling::SPECIAL_HANDLING, + FilterlessUpdate, bind)); } static void AddAvgOverloads(AggregateFunctionSet &set, const string &name, aggregate_finalize_t finalize, @@ -1237,11 +1361,66 @@ static void AddAvgOverloads(AggregateFunctionSet &set, const string &name, aggre set.AddFunction(AggregateFunction( name, {LogicalType::UBIGINT, LogicalType::BOOLEAN, LogicalType::DOUBLE, LogicalType::DOUBLE, LogicalType::DOUBLE, - LogicalType::DOUBLE, LogicalType::DOUBLE, LogicalType::DOUBLE, LogicalType::UBIGINT}, + LogicalType::DOUBLE, LogicalType::DOUBLE, LogicalType::DOUBLE}, return_type, FilterlessAvgStateSize, FilterlessAvgInitialize, FilterlessAvgScatterUpdate, FilterlessAvgCombine, finalize, FunctionNullHandling::SPECIAL_HANDLING, FilterlessAvgUpdate, BindFilterlessAvg)); } +template +static AggregateFunction MakeFilterlessFusedAvgFunction(const LogicalType &sum_type, bind_aggregate_function_t bind) { + auto function = + AggregateFunction("priv_filterless_avg", + {LogicalType::UBIGINT, LogicalType::BOOLEAN, sum_type, LogicalType::BIGINT, sum_type, + LogicalType::BIGINT, LogicalType::DOUBLE, LogicalType::DOUBLE}, + LogicalType::DOUBLE, FilterlessFusedAvgStateSize, + FilterlessFusedAvgInitialize, FilterlessFusedAvgScatterUpdate, + FilterlessFusedAvgCombine, FilterlessFusedAvgFinalize, + FunctionNullHandling::SPECIAL_HANDLING, FilterlessFusedAvgUpdate, bind); + function.SetOrderDependent(AggregateOrderDependent::NOT_ORDER_DEPENDENT); + return function; +} + +static unique_ptr BindFilterlessFusedApproxAvg(ClientContext &context, AggregateFunction &, + vector> &arguments) { + return BindFilterless(context, arguments, 6, true, FILTERLESS_HUGEINT_BIN_COUNT); +} + +static unique_ptr BindFilterlessFusedExactAvg(ClientContext &context, AggregateFunction &, + vector> &arguments) { + return BindFilterless(context, arguments, 6, false, FILTERLESS_HUGEINT_BIN_COUNT); +} + +static unique_ptr BindFilterlessFusedDecimalAvg(ClientContext &context, AggregateFunction &function, + vector> &arguments) { + auto sum_type = arguments[2]->return_type; + if (arguments[4]->return_type != sum_type) { + throw InvalidInputException("priv_filterless_avg: answer and histogram DECIMAL types must match"); + } + switch (sum_type.InternalType()) { + case PhysicalType::INT16: + function = MakeFilterlessFusedAvgFunction>( + sum_type, BindFilterlessFusedDecimalAvg); + break; + case PhysicalType::INT32: + function = MakeFilterlessFusedAvgFunction>( + sum_type, BindFilterlessFusedDecimalAvg); + break; + case PhysicalType::INT64: + function = MakeFilterlessFusedAvgFunction>( + sum_type, BindFilterlessFusedDecimalAvg); + break; + case PhysicalType::INT128: + function = MakeFilterlessFusedAvgFunction>( + sum_type, BindFilterlessFusedDecimalAvg); + break; + default: + throw InternalException("priv_filterless_avg: unsupported DECIMAL physical type"); + } + auto result = BindFilterless(context, arguments, 6, false, FILTERLESS_HUGEINT_BIN_COUNT); + ConfigureFilterlessDecimalBind(result->Cast(), sum_type); + return result; +} + template static AggregateFunction MakeFilterlessLowerPairFunction(const string &name, const LogicalType &input_type, const LogicalType &return_type) { @@ -1286,6 +1465,22 @@ static unique_ptr BindFilterlessDecimalSumPair(ClientContext &, Ag } void RegisterFilterlessAggregateFunctions(ExtensionLoader &loader) { + AggregateFunctionSet fused_avg_set("priv_filterless_avg"); + fused_avg_set.AddFunction(MakeFilterlessFusedAvgFunction( + LogicalType::DOUBLE, BindFilterlessFusedApproxAvg)); + fused_avg_set.AddFunction(MakeFilterlessFusedAvgFunction>( + LogicalType::HUGEINT, BindFilterlessFusedExactAvg)); + fused_avg_set.AddFunction( + AggregateFunction({LogicalType::UBIGINT, LogicalType::BOOLEAN, LogicalTypeId::DECIMAL, LogicalType::BIGINT, + LogicalTypeId::DECIMAL, LogicalType::BIGINT, LogicalType::DOUBLE, LogicalType::DOUBLE}, + LogicalType::DOUBLE, nullptr, nullptr, nullptr, nullptr, nullptr, + FunctionNullHandling::SPECIAL_HANDLING, nullptr, BindFilterlessFusedDecimalAvg)); + CreateAggregateFunctionInfo fused_avg_info(fused_avg_set); + FunctionDescription fused_avg_description; + fused_avg_description.description = "[INTERNAL] Fused filterless SUM and exact COUNT components for AVG."; + fused_avg_info.descriptions.push_back(std::move(fused_avg_description)); + loader.RegisterFunction(std::move(fused_avg_info)); + AggregateFunction count_pair = MakeFilterlessLowerPairFunction( "priv_filterless_count_pair", LogicalType::BOOLEAN, LogicalType::BIGINT); CreateAggregateFunctionInfo count_pair_info(count_pair); @@ -1324,11 +1519,10 @@ void RegisterFilterlessAggregateFunctions(ExtensionLoader &loader) { AggregateFunction({LogicalType::UBIGINT, LogicalType::BOOLEAN, LogicalTypeId::DECIMAL, LogicalTypeId::DECIMAL}, LogicalTypeId::DECIMAL, nullptr, nullptr, nullptr, nullptr, nullptr, FunctionNullHandling::SPECIAL_HANDLING, nullptr, BindFilterlessDecimalSum)); - sum_set.AddFunction( - AggregateFunction({LogicalType::UBIGINT, LogicalType::BOOLEAN, LogicalTypeId::DECIMAL, LogicalTypeId::DECIMAL, - LogicalType::DOUBLE, LogicalType::DOUBLE, LogicalType::UBIGINT}, - LogicalTypeId::DECIMAL, nullptr, nullptr, nullptr, nullptr, nullptr, - FunctionNullHandling::SPECIAL_HANDLING, nullptr, BindFilterlessDecimalSum)); + sum_set.AddFunction(AggregateFunction({LogicalType::UBIGINT, LogicalType::BOOLEAN, LogicalTypeId::DECIMAL, + LogicalTypeId::DECIMAL, LogicalType::DOUBLE, LogicalType::DOUBLE}, + LogicalTypeId::DECIMAL, nullptr, nullptr, nullptr, nullptr, nullptr, + FunctionNullHandling::SPECIAL_HANDLING, nullptr, BindFilterlessDecimalSum)); CreateAggregateFunctionInfo sum_info(sum_set); FunctionDescription sum_description; sum_description.description = diff --git a/src/compiler/privacy_mechanisms.cpp b/src/compiler/privacy_mechanisms.cpp index 7786ea1..9bfa294 100644 --- a/src/compiler/privacy_mechanisms.cpp +++ b/src/compiler/privacy_mechanisms.cpp @@ -3291,36 +3291,6 @@ static FilterlessPreAggregationInput ApplyFilterlessMaxGroups(OptimizerExtension return result; } -static unique_ptr BuildFilterlessNonce(OptimizerExtensionInput &input, - const FilterlessPreAggregationInput &pre_input, - idx_t component_index, uint64_t query_nonce) { - unique_ptr nonce = make_uniq( - Value::UBIGINT(PAC_MAGIC_HASH ^ static_cast(component_index + 1) ^ query_nonce)); - for (idx_t i = 0; i < pre_input.group_count; i++) { - auto group_hash = input.optimizer.BindScalarFunction("hash", pre_input.GroupRef(i)); - nonce = input.optimizer.BindScalarFunction("xor", std::move(nonce), std::move(group_hash)); - } - return nonce; -} - -static uint64_t FilterlessQueryNonce(const string &query_hash) { - uint64_t result = 0; - for (auto character : query_hash) { - uint64_t digit; - if (character >= '0' && character <= '9') { - digit = static_cast(character - '0'); - } else if (character >= 'a' && character <= 'f') { - digit = static_cast(character - 'a' + 10); - } else if (character >= 'A' && character <= 'F') { - digit = static_cast(character - 'A' + 10); - } else { - throw InternalException("dp_filterless: normalized query hash is not hexadecimal"); - } - result = (result << 4) | digit; - } - return result; -} - static unique_ptr BuildFilterlessActiveSupportKey(OptimizerExtensionInput &input, const FilterlessPreAggregationInput &pre_input) { auto active_count = pre_input.ActiveCountRef(); @@ -3368,11 +3338,9 @@ static unique_ptr BuildFilterlessLowerPair(OptimizerExtensionInput & } void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, OptimizerExtensionInput &input, - unique_ptr &plan, const vector &privacy_units, - const string &query_hash) { + unique_ptr &plan, const vector &privacy_units, const string &) { PRIVACY_DEBUG_PRINT("[dp_filterless] CompileDPFilterlessQuery: start"); auto filterless_settings = GetFilterlessSettings(input.context); - uint64_t query_nonce = FilterlessQueryNonce(query_hash); double epsilon = GetValidatedDpEpsilon(input.context, "dp_filterless"); bool allow_self_joins = ValidateDPSelfJoins(plan, "dp_filterless") > 1.0; auto chain = ExtractDPFKChain(plan, privacy_units, check, allow_self_joins); @@ -3419,15 +3387,13 @@ void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, Optimizer auto avg_infos = RewriteAvgAggregates(input, agg, AvgRewriteMode::PLAIN, nullptr, /*count_non_null_value=*/true); idx_t component_count = agg->expressions.size(); - auto avg_components = BuildAvgComponentSet(avg_infos); agg->ResolveOperatorTypes(); vector component_output_types; - component_output_types.reserve(component_count); - for (idx_t i = 0; i < component_count; i++) { - // AVG's internal SUM and COUNT remain DOUBLE until the ratio is formed, so - // Laplace noise is not truncated by an intermediate integer cast. - component_output_types.push_back(avg_components.count(i) ? LogicalType::DOUBLE : agg->types[n_groups + i]); + component_output_types.reserve(original_aggregate_count); + for (idx_t i = 0; i < original_aggregate_count; i++) { + component_output_types.push_back(FindAvgInfoForSumPos(avg_infos, i) ? LogicalType::DOUBLE + : agg->types[n_groups + i]); } vector count_components; @@ -3451,27 +3417,39 @@ void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, Optimizer "contributions; floating SUM uses the scalar AS magnitude accumulator"); auto pre_input = ApplyFilterlessMaxGroups(input, agg, pre, component_count, max_groups); double visible_cell_epsilon = epsilon / budget_units; - for (idx_t i = 0; i < component_count; i++) { - double component_epsilon = avg_components.count(i) ? visible_cell_epsilon / 2.0 : visible_cell_epsilon; - unique_ptr answer_partial = pre_input.AnswerRef(input, i); - unique_ptr histogram_partial = pre_input.HistogramRef(input, i); + vector> upper_expressions; + upper_expressions.reserve(original_aggregate_count + (n_groups > 0 ? 1 : 0)); + for (idx_t i = 0; i < original_aggregate_count; i++) { auto is_active = make_uniq(ExpressionType::COMPARE_GREATERTHAN, pre_input.ActiveCountRef(), make_uniq(Value::BIGINT(0))); vector> children; children.push_back(pre_input.PuRef()); children.push_back(std::move(is_active)); - children.push_back(std::move(answer_partial)); - children.push_back(std::move(histogram_partial)); - children.push_back(make_uniq(Value::DOUBLE(component_epsilon))); + auto *avg_info = FindAvgInfoForSumPos(avg_infos, i); + if (avg_info) { + children.push_back(pre_input.AnswerRef(input, avg_info->sum_pos)); + children.push_back(pre_input.AnswerRef(input, avg_info->count_pos)); + children.push_back(pre_input.HistogramRef(input, avg_info->sum_pos)); + children.push_back(pre_input.HistogramRef(input, avg_info->count_pos)); + children.push_back(make_uniq(Value::DOUBLE(visible_cell_epsilon))); + children.push_back(make_uniq(Value::DOUBLE(static_cast(max_groups)))); + upper_expressions.push_back(BindAggregateLocal(input, "priv_filterless_avg", std::move(children))); + PRIVACY_DEBUG_PRINT("[dp_filterless] fused upper AVG " + std::to_string(i) + + " epsilon=" + std::to_string(visible_cell_epsilon)); + continue; + } + children.push_back(pre_input.AnswerRef(input, i)); + children.push_back(pre_input.HistogramRef(input, i)); + children.push_back(make_uniq(Value::DOUBLE(visible_cell_epsilon))); children.push_back(make_uniq(Value::DOUBLE(static_cast(max_groups)))); - children.push_back(BuildFilterlessNonce(input, pre_input, i, query_nonce)); - agg->expressions[i] = - BindAggregateLocal(input, count_components[i] ? "filterless_count" : "filterless_sum", std::move(children)); + upper_expressions.push_back(BindAggregateLocal( + input, count_components[i] ? "filterless_count" : "filterless_sum", std::move(children))); PRIVACY_DEBUG_PRINT("[dp_filterless] component " + std::to_string(i) + (count_components[i] ? " COUNT" : " SUM") + - " epsilon=" + std::to_string(component_epsilon)); + " epsilon=" + std::to_string(visible_cell_epsilon)); } + agg->expressions = std::move(upper_expressions); optional_idx support_pos; LogicalOperator *projection_anchor = agg; if (n_groups > 0) { @@ -3486,13 +3464,8 @@ void CompileDPFilterlessQuery(const PrivacyCompatibilityResult &check, Optimizer // Normalize the custom aggregate outputs onto one projection table and restore the // component types expected by operators above the original aggregate. - vector zero_scales(component_count, 0.0); - auto output_projection = - WrapAggregateWithLaplace(input, plan, agg, projection_anchor, zero_scales, component_output_types); - if (!avg_infos.empty()) { - WrapAvgRatioProjection(input, plan, output_projection, avg_infos, n_groups, original_aggregate_count, - output_projection.proj_ptr, /*null_on_nonpositive_count=*/true); - } + vector zero_scales(original_aggregate_count, 0.0); + WrapAggregateWithLaplace(input, plan, agg, projection_anchor, zero_scales, component_output_types); #if PRIVACY_DEBUG PRIVACY_DEBUG_PRINT("=== PLAN AFTER dp_filterless TRANSFORMATION ==="); diff --git a/test/sql/dp_filterless.test b/test/sql/dp_filterless.test index 5df7ee9..009de2a 100644 --- a/test/sql/dp_filterless.test +++ b/test/sql/dp_filterless.test @@ -137,29 +137,18 @@ ORDER BY world; 0 128.000000 100.000000 1 128.000000 228.000000 -# Explicit nonces are group identities supplied by the compiler. Accepting two -# values in one aggregate group would make the release depend on row order. -statement error -SELECT filterless_sum(pu, true, contribution, contribution, 1.0, 1.0, nonce) -FROM (VALUES - (1::UBIGINT, 50.0::DOUBLE, 1::UBIGINT), - (3::UBIGINT, 50.0::DOUBLE, 2::UBIGINT) -) input(pu, contribution, nonce); ----- -filterless: noise nonce must be constant within each aggregate group - statement ok SET dp_filterless_sample_bits = 0; statement ok SET dp_filterless_clip_support = 2.0; -# Exercise parallel aggregate combine, including the explicit nonce invariant. +# Exercise parallel aggregate combine with explicit epsilon and C_u. statement ok SET threads = 4; query R -SELECT filterless_sum((i::UBIGINT << 1), true, 50.0, 50.0, 100.0, 1.0, 7::UBIGINT) +SELECT filterless_sum((i::UBIGINT << 1), true, 50.0, 50.0, 100.0, 1.0) FROM range(100000) t(i); ---- 5000000.000000 @@ -198,8 +187,9 @@ FROM filterless_people; ---- 228.000000 3 76.000000 -# Equivalent populations reached through different predicates use different -# query nonces. Otherwise sticky noise would cancel across a filter attack. +# Equivalent populations reached through different predicates receive fresh +# secure noise. The filterless aggregate does not need a query nonce because +# its release is intentionally non-sticky. statement ok SET dp_epsilon = 1.0; @@ -544,6 +534,24 @@ WHERE qualifies; ---- 10000.000000 2 4 5000.000000 +# AVG keeps its SUM and exact COUNT components in one upper aggregate. The +# lower paired partials remain separate because they are formed per logical PU. +query II +EXPLAIN SELECT AVG(amount) +FROM filterless_sampled_events +WHERE qualifies; +---- +physical_plan :[\s\S]*UNGROUPED_AGGREGATE[\s\S]*priv_filterless_avg[\s\S]*priv_filterless_sum_pair[\s\S]*priv_filterless_count_pair[\s\S]* + +# COUNT(expr) is zero when every qualifying value is NULL, so fused AVG keeps +# SQL's NULL result instead of dividing by a synthetic denominator. +query I +SELECT AVG(amount) IS NULL +FROM filterless_sampled_events +WHERE amount IS NULL; +---- +true + statement ok SET dp_filterless_sample_bits = 0; @@ -605,6 +613,15 @@ FROM filterless_exact_numeric; ---- 9007199254740995 90071992547409.95 HUGEINT DECIMAL(38,2) +# The fused AVG accepts the exact HUGEINT/DECIMAL lower partials and only +# converts the released ratio to AVG's public DOUBLE result. +query RRTT +SELECT AVG(integer_amount), AVG(decimal_amount), + typeof(AVG(integer_amount)), typeof(AVG(decimal_amount)) +FROM filterless_exact_numeric; +---- +4503599627370498.000000 45035996273704.976562 DOUBLE DOUBLE + # The public SQL aggregate overload uses the same exact state as the compiler. query RT SELECT filterless_sum(pu, true, answer_value, histogram_value),