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 d44e054..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]; } @@ -482,27 +472,31 @@ 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); } 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++; } } @@ -527,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(); @@ -540,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) { @@ -564,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)}; } @@ -604,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); @@ -632,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 { @@ -664,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 @@ -686,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); } } @@ -755,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); - } } } @@ -767,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); - } } } @@ -778,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); - } } } @@ -822,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 { @@ -840,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) @@ -896,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) { @@ -903,43 +875,190 @@ 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)); } } -// 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_t = double; + using state_t = FilterlessApproxSumState; + using result_t = double; static hugeint_t ApproximateScaledValue(double value) { return Hugeint::Convert(ClipApproximateMagnitude64(AsScaledMagnitude(value))); } - template - static void Operation(STATE &state, const INPUT_TYPE &input, AggregateUnaryInput &) { - double value = static_cast(input); + static void Add(state_t &state, input_t value) { if (!std::isfinite(value)) { throw InvalidInputException("filterless: per-PU SUM contribution must be finite"); } @@ -952,43 +1071,174 @@ struct FilterlessApproxSumOperation { } } - 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_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); } - template - static void Finalize(STATE &state, RESULT_TYPE &target, AggregateFinalizeData &finalize_data) { - if (!state.isset) { - finalize_data.ReturnNull(); - return; - } + static bool IsSet(const state_t &state) { + return state.isset; + } + + static result_t Finalize(const state_t &state) { auto scaled = Hugeint::Subtract(state.positive, state.negative); - target = Hugeint::Cast(scaled) / CLIP_DOUBLE_SCALE; + return Hugeint::Cast(scaled) / CLIP_DOUBLE_SCALE; + } +}; + +struct FilterlessExactSumPartialState { + bool isset; + hugeint_t value; +}; + +template +struct FilterlessLowerPairState { + PARTIAL_STATE answer; + PARTIAL_STATE histogram; +}; + +struct FilterlessCountPairOperation { + using input_t = bool; + using state_t = uint64_t; + using result_t = int64_t; + + static void Add(state_t &state, input_t value) { + state += static_cast(value); + } + + static void Combine(const state_t &source, state_t &target) { + target += source; } - static bool IgnoreNull() { + static bool IsSet(const state_t &) { return true; } + + static result_t Finalize(const state_t &state) { + return static_cast(state); + } }; +template +struct FilterlessExactSumPairOperation { + using input_t = INPUT; + using state_t = FilterlessExactSumPartialState; + using result_t = hugeint_t; + + 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_t &source, state_t &target) { + target.isset = target.isset || source.isset; + target.value = Hugeint::Add(target.value, source.value); + } + + static bool IsSet(const state_t &state) { + return state.isset; + } + + static result_t Finalize(const state_t &state) { + return state.value; + } +}; + +static LogicalType FilterlessLowerPairType(const LogicalType &value_type) { + return LogicalType::STRUCT({{"answer", value_type}, {"histogram", value_type}}); +} + +template +static idx_t FilterlessLowerPairStateSize(const AggregateFunction &) { + return sizeof(FilterlessLowerPairState); +} + +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) { + 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 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) { + 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]); + } + } + } +} + +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; }); +} + +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)]; }); +} + +template +static void FilterlessLowerPairCombine(Vector &source, Vector &target, AggregateInputData &, idx_t count) { + 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); + } +} + +template +static void FilterlessLowerPairFinalize(Vector &states, AggregateInputData &, Vector &result, idx_t count, + idx_t offset) { + 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]); + for (idx_t i = 0; i < count; i++) { + auto row = offset + i; + if (OPERATION::IsSet(state_ptrs[i]->answer)) { + answers[row] = OPERATION::Finalize(state_ptrs[i]->answer); + } else { + FlatVector::Validity(*children[0]).SetInvalid(row); + } + if (OPERATION::IsSet(state_ptrs[i]->histogram)) { + histograms[row] = OPERATION::Finalize(state_ptrs[i]->histogram); + } else { + FlatVector::Validity(*children[1]).SetInvalid(row); + } + } +} + static LogicalType FilterlessDebugType() { child_list_t children; children.emplace_back("lower_bound", LogicalType::DOUBLE); @@ -1029,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, @@ -1065,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; @@ -1075,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; } @@ -1090,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, @@ -1109,22 +1361,152 @@ 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) { + 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) { + return MakeFilterlessLowerPairFunction>("priv_filterless_sum_pair", + input_type, return_type); +} + +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()) { + 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_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)); + 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); + 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_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, 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, @@ -1137,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 280b777..9bfa294 100644 --- a/src/compiler/privacy_mechanisms.cpp +++ b/src/compiler/privacy_mechanisms.cpp @@ -3188,16 +3188,23 @@ 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 { + vector> children; + children.push_back(AggregateRef(index)); + children.push_back(make_uniq(Value(answer ? "answer" : "histogram"))); + return BindScalarLocal(input, "struct_extract", std::move(children)); + } + + unique_ptr AnswerRef(OptimizerExtensionInput &input, idx_t index) const { + return ComponentRef(input, index, true); } - unique_ptr HistogramRef(idx_t index) const { - return AggregateRef(component_count + index); + 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(component_count); } }; @@ -3212,7 +3219,7 @@ static FilterlessPreAggregationInput BuildFilterlessPreAggregationInput(const Pu 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 < component_count + 1; i++) { result.aggregate_types.push_back(pre.lower_agg->types[pre.num_original_groups + 1 + i]); } return result; @@ -3229,7 +3236,7 @@ static FilterlessPreAggregationInput ApplyFilterlessMaxGroups(OptimizerExtension // 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 = component_count + 1; idx_t projection_index = input.optimizer.binder.GenerateTableIndex(); vector> expressions; expressions.reserve(pre.num_original_groups + 1 + lower_aggregate_count); @@ -3284,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(); @@ -3341,27 +3318,29 @@ 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 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()) { - return BindPlainAggregate(input, "count", aggregate.children[0]->Copy()); + children.push_back(BuildIsNotNullPredicate(aggregate.children[0]->Copy())); + } else { + children.push_back(make_uniq(Value::BOOLEAN(true))); } - return BindPlainAggregate(input, "count_star", nullptr); - } - 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()); + } 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_sum_pair", + std::move(children)); } 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); @@ -3408,37 +3387,25 @@ 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; vector> lower_expressions; count_components.reserve(component_count); - lower_expressions.reserve(2 * component_count + 1); - vector> histogram_expressions; - histogram_expressions.reserve(component_count); + lower_expressions.reserve(component_count + 1); 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); - answer_partial->Cast().filter = BuildFilterlessIsActive(input, *encoded_pu); - lower_expressions.push_back(std::move(answer_partial)); - auto histogram_partial = BuildFilterlessLowerAggregate(input, aggregate, is_count); - 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( + BuildFilterlessLowerPair(input, aggregate, is_count, *encoded_pu, filterless_settings.sample_bits)); } auto active_count = BindPlainAggregate(input, "count_star", nullptr); active_count->Cast().filter = BuildFilterlessIsActive(input, *encoded_pu); @@ -3446,31 +3413,43 @@ 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 separate filtered-answer and fixed-sample histogram " + 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, 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(i); - unique_ptr histogram_partial = pre_input.HistogramRef(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) { @@ -3485,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 de06e1c..009de2a 100644 --- a/test/sql/dp_filterless.test +++ b/test/sql/dp_filterless.test @@ -77,6 +77,23 @@ FROM ( ---- 25.000000 100.000000 4.000000 +# 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_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) +); +---- +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. @@ -120,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 @@ -181,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; @@ -501,6 +508,50 @@ WHERE NOT qualifies; ---- 2000.000000 2 1000.000000 +# 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'; + +query II +EXPLAIN SELECT SUM(amount), COUNT(*) +FROM filterless_sampled_events +WHERE qualifies; +---- +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. +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 + +# 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; @@ -549,6 +600,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_sum_pair[\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)) @@ -556,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),