From 12f35cd39b8a1953296ff365ca8c82b020b38969 Mon Sep 17 00:00:00 2001 From: ila Date: Tue, 1 Sep 2026 22:07:03 +0200 Subject: [PATCH] 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),