From 37e6de72227654ffac905c85a8d901326e3ca983 Mon Sep 17 00:00:00 2001
From: Phil Haack <haacked@gmail.com>
Date: Tue, 15 Sep 2026 14:57:45 -0700
Subject: [PATCH 1/7] fix(flags): bound the release condition size query's
 memory

Size a person release condition from a sample of the person keyspace,
behind flags-blast-radius-query-v2, so the count stops crossing the
per-query memory limit on large projects.

The sampling and memory-bounded settings move out of the workflows
audience service into products/feature_flags, which both products now
share.

Generated-By: PostHog Desktop
Task-Id: be8e214b-4146-4efa-8f3c-5bd3bd59210a
---
 .../feature_flags/backend/person_sampling.py  | 142 ++++++++++++++++
 .../backend/test/test_blast_radius_v2.py      |  53 ++++++
 .../backend/user_blast_radius.py              |  46 ++++++
 products/workflows/backend/api/hog_flow.py    |   2 +-
 .../workflows/backend/services/audience_v2.py | 156 +++---------------
 .../backend/test/test_audience_v2.py          |  15 +-
 6 files changed, 268 insertions(+), 146 deletions(-)
 create mode 100644 products/feature_flags/backend/person_sampling.py
 create mode 100644 products/feature_flags/backend/test/test_blast_radius_v2.py

diff --git a/products/feature_flags/backend/person_sampling.py b/products/feature_flags/backend/person_sampling.py
new file mode 100644
index 000000000000..fa0f4b290e40
--- /dev/null
+++ b/products/feature_flags/backend/person_sampling.py
@@ -0,0 +1,142 @@
+from collections.abc import Callable
+from typing import Optional
+
+from posthog.schema import HogQLQueryModifiers
+
+from posthog.hogql import ast
+from posthog.hogql.constants import HogQLGlobalSettings
+from posthog.hogql.context import HogQLContext
+from posthog.hogql.database.database import Database
+from posthog.hogql.property import property_to_expr
+from posthog.hogql.query import execute_hogql_query
+
+from posthog.models.filters import Filter
+from posthog.models.team.team import Team
+from posthog.schema_enums import PersonsArgMaxVersion
+
+# 1-in-64 sample of the person keyspace. Sampling keys on person id (the dedup key), so every
+# version row of a sampled person lands in the sample and the argMax dedup stays exact within it.
+SAMPLE_MODULUS = 64
+
+# Below this many sampled matches the extrapolation is too noisy, so rerun exact. The relative
+# error of the estimate is about sqrt(63 / matched_persons), so 10,000 sampled matches
+# (~640k matched persons) keeps the worst case near 3%. The exact query is usually cheap in
+# that regime: few matching persons means the id prefilter keeps the dedup aggregation small.
+# Caveat: the prefilter matches any historical row version, so a churny property (many persons
+# matched once, few match now) can make the exact rerun carry a candidate set far larger than
+# the current match count suggests.
+MIN_SAMPLED_MATCHES = 10_000
+
+
+def bounded_memory_settings() -> HogQLGlobalSettings:
+    """
+    Execution settings that keep person-dedup aggregations memory-bounded on large teams.
+
+    The person table sorts by (team_id, id), so with in-order aggregation the dedup
+    GROUP BY id streams instead of holding every person in a hash table. The spill
+    threshold is the backstop for aggregations where in-order cannot apply (for
+    example cohort subqueries): degrade to disk instead of a memory-limit error.
+    """
+    return HogQLGlobalSettings(
+        optimize_aggregation_in_order=True,
+        max_bytes_before_external_group_by=4 * 1024**3,
+    )
+
+
+def count_settings(sample_modulus: Optional[int]) -> HogQLGlobalSettings:
+    # A sampled count keeps the fast parallel hash aggregation: the sample already bounds
+    # the hash table, and in-order aggregation only makes it slower. The exact runs need
+    # the streaming mode to stay memory-bounded on large teams.
+    if sample_modulus is not None:
+        return HogQLGlobalSettings(max_bytes_before_external_group_by=4 * 1024**3)
+    return bounded_memory_settings()
+
+
+def sampled_or_exact_count(run_count: Callable[[Optional[int]], int]) -> int:
+    """
+    Count from a 1-in-SAMPLE_MODULUS sample and extrapolate, or exactly when the sample holds
+    too few matches for a stable extrapolation.
+
+    `run_count` takes the sample modulus, or None for an exact run.
+    """
+    sampled = run_count(SAMPLE_MODULUS)
+    if sampled >= MIN_SAMPLED_MATCHES:
+        return sampled * SAMPLE_MODULUS
+    return run_count(None)
+
+
+def count_matching_persons(team: Team, filter: Optional[Filter], database: Database, query_type: str) -> int:
+    """Count the persons a filter matches, or every person on the team when filter is None."""
+    return sampled_or_exact_count(
+        lambda sample_modulus: _run_person_count(team, filter, database, query_type, sample_modulus)
+    )
+
+
+def _run_person_count(
+    team: Team, filter: Optional[Filter], database: Database, query_type: str, sample_modulus: Optional[int]
+) -> int:
+    query = build_person_count_query(team, filter, sample_modulus=sample_modulus)
+    response = execute_hogql_query(
+        query=query,
+        team=team,
+        query_type=query_type,
+        # Pin v1 for the unfiltered count, as count_persons_seen_so_far does: the v2 persons
+        # path pushes the executor's default LIMIT into its dedup subquery, which caps a bare
+        # count() at ~101 for teams pinned to v2 (see #87323). A filtered count reads person
+        # properties, where v2 is the faster shape, so it keeps the automatic choice.
+        modifiers=HogQLQueryModifiers(personsArgMaxVersion=PersonsArgMaxVersion.V1) if filter is None else None,
+        context=HogQLContext(team_id=team.pk, database=database),
+        settings=count_settings(sample_modulus),
+    )
+    return response.results[0][0] if response.results else 0
+
+
+def build_person_count_query(team: Team, filter: Optional[Filter], sample_modulus: Optional[int]) -> ast.SelectQuery:
+    where_exprs: list[ast.Expr] = [
+        ast.CompareOperation(
+            op=ast.CompareOperationOp.Eq,
+            left=ast.Field(chain=["persons", "team_id"]),
+            right=ast.Constant(value=team.pk),
+        )
+    ]
+    if sample_modulus is not None:
+        where_exprs.append(person_sample_predicate(sample_modulus))
+    if filter is not None:
+        where_exprs.append(property_to_expr(filter.property_groups, team, scope="person"))
+
+    # A filter can add a one-to-many join: a `distinct_id` person property resolves through
+    # persons.pdi, which gives a person one row per distinct id. So a filtered count dedups on
+    # the person id. The unfiltered total joins nothing, so it keeps the plain count() and
+    # avoids a uniqExact state over every person on the team.
+    if filter is None:
+        count_expr: ast.Expr = ast.Call(name="count", args=[])
+    else:
+        count_expr = ast.Call(name="count", distinct=True, args=[ast.Field(chain=["persons", "id"])])
+
+    return ast.SelectQuery(
+        select=[count_expr],
+        select_from=ast.JoinExpr(table=ast.Field(chain=["persons"])),
+        where=ast.And(exprs=where_exprs),
+    )
+
+
+def person_sample_predicate(modulus: int) -> ast.Expr:
+    # Built as modulo(...) calls, not the % operator: WhereClauseExtractor fails safe to
+    # "no prefilter" on ArithmeticOperation nodes, and the memory bound depends on this
+    # predicate reaching the raw person scan inside the persons lazy table.
+    return sample_predicate(ast.Field(chain=["persons", "id"]), modulus)
+
+
+def sample_predicate(key: ast.Expr, modulus: int) -> ast.Expr:
+    """Keep 1 in `modulus` of the values `key` takes, spread evenly by a hash of the key."""
+    return ast.CompareOperation(
+        op=ast.CompareOperationOp.Eq,
+        left=ast.Call(
+            name="modulo",
+            args=[
+                ast.Call(name="cityHash64", args=[key]),
+                ast.Constant(value=modulus),
+            ],
+        ),
+        right=ast.Constant(value=0),
+    )
diff --git a/products/feature_flags/backend/test/test_blast_radius_v2.py b/products/feature_flags/backend/test/test_blast_radius_v2.py
new file mode 100644
index 000000000000..0fc1723b3987
--- /dev/null
+++ b/products/feature_flags/backend/test/test_blast_radius_v2.py
@@ -0,0 +1,53 @@
+from posthog.test.base import BaseTest, ClickhouseTestMixin, _create_person, flush_persons_and_events
+from unittest.mock import patch
+
+from parameterized import parameterized
+
+from products.feature_flags.backend.user_blast_radius import get_user_blast_radius
+
+FILTERS = {"properties": [{"key": "subscribed", "type": "person", "value": ["true"], "operator": "exact"}]}
+
+GATE = "products.feature_flags.backend.user_blast_radius.use_blast_radius_query_v2"
+
+
+class TestBlastRadiusQueryV2(ClickhouseTestMixin, BaseTest):
+    def _create_persons(self, subscribed_flags: list[bool]) -> None:
+        for i, subscribed in enumerate(subscribed_flags, start=1):
+            _create_person(
+                team=self.team,
+                distinct_ids=[f"user-{i}"],
+                properties={"subscribed": "true" if subscribed else "false"},
+            )
+        flush_persons_and_events()
+
+    @parameterized.expand(
+        [
+            ("matching a person property", FILTERS, 3, 4),
+            ("with no conditions", {}, 4, 4),
+        ]
+    )
+    def test_sized_condition_matches_the_v1_counts(self, _name, filters, affected, total):
+        # A handful of persons never clears MIN_SAMPLED_MATCHES under the default sample, so
+        # this exercises the exact fallback and pins both counts to the v1 result.
+        self._create_persons([True, True, True, False])
+
+        v1_result = get_user_blast_radius(self.team, filters)
+        with patch(GATE, return_value=True):
+            result = get_user_blast_radius(self.team, filters)
+
+        assert (result.affected, result.total) == (affected, total)
+        assert (result.affected, result.total) == (v1_result.affected, v1_result.total)
+
+    def test_sampled_path_extrapolates_by_modulus(self):
+        self._create_persons([True, True, False])
+
+        # Modulus 1 samples everyone and MIN_SAMPLED_MATCHES 0 forces the sampled branch,
+        # so the extrapolated result must equal the exact count.
+        with (
+            patch(GATE, return_value=True),
+            patch("products.feature_flags.backend.person_sampling.SAMPLE_MODULUS", 1),
+            patch("products.feature_flags.backend.person_sampling.MIN_SAMPLED_MATCHES", 0),
+        ):
+            result = get_user_blast_radius(self.team, FILTERS)
+
+        assert (result.affected, result.total) == (2, 3)
diff --git a/products/feature_flags/backend/user_blast_radius.py b/products/feature_flags/backend/user_blast_radius.py
index 5dccd047f48c..9c73ca9dcc6f 100644
--- a/products/feature_flags/backend/user_blast_radius.py
+++ b/products/feature_flags/backend/user_blast_radius.py
@@ -4,6 +4,7 @@
 
 from django.core.exceptions import ObjectDoesNotExist
 
+import posthoganalytics
 from rest_framework.exceptions import ValidationError
 
 from posthog.schema import PropertyOperator
@@ -28,6 +29,7 @@
 from posthog.models.team.team import Team
 
 from products.cohorts.backend.models.cohort import Cohort
+from products.feature_flags.backend.person_sampling import bounded_memory_settings, count_matching_persons
 
 
 @frozen
@@ -36,6 +38,43 @@ class BlastRadiusResult:
     total: int
 
 
+BLAST_RADIUS_QUERY_V2_FLAG = "flags-blast-radius-query-v2"
+
+QUERY_TYPE_V2 = "feature_flag_blast_radius_v2"
+
+
+def use_blast_radius_query_v2(team: Team) -> bool:
+    return bool(
+        posthoganalytics.feature_enabled(
+            BLAST_RADIUS_QUERY_V2_FLAG,
+            str(team.uuid),
+            send_feature_flag_events=False,
+        )
+    )
+
+
+def sampled_person_blast_radius(team: Team, filter: Filter, query_type: str) -> BlastRadiusResult:
+    """
+    Person blast radius whose peak query memory does not grow with the size of the person table.
+
+    The exact counts dedup the person table with a hash GROUP BY that holds every matched
+    person in memory, and HogQL queries cannot spill to disk, so on a large team both counts
+    cross the per-query memory limit and the caller gets an error instead of a number. These
+    counts read a sample of the persons and extrapolate, and fall back to an exact, streaming
+    count when the sample holds too few matches to extrapolate from.
+    """
+    # One database build shared by both counts; each execute_hogql_query call would otherwise
+    # rebuild it, and the build cost scales with the team's warehouse size.
+    database = Database.create_for(team=team)
+
+    total = count_matching_persons(team, None, database, query_type=query_type)
+    if len(filter.property_groups.flat) == 0:
+        return BlastRadiusResult(affected=total, total=total)
+
+    affected = count_matching_persons(team, filter, database, query_type=query_type)
+    return BlastRadiusResult(affected=min(affected, total), total=total)
+
+
 # ClickHouse codes for "this literal can't be parsed as the column's type": 6 CANNOT_PARSE_TEXT,
 # 72 CANNOT_PARSE_NUMBER — e.g. a numeric operator (gt/lt) against a null or non-numeric filter
 # value casts 'None' to Float64 and fails deterministically. Both are classified USER_ERROR in
@@ -149,6 +188,10 @@ def get_user_blast_radius_persons(
 def _get_person_blast_radius(team: Team, filter: Filter) -> BlastRadiusResult:
     """Calculate blast radius for person-based feature flags using HogQL."""
 
+    if use_blast_radius_query_v2(team):
+        tag_queries(product=Product.FEATURE_FLAGS, feature=Feature.QUERY)
+        return sampled_person_blast_radius(team, filter, query_type=QUERY_TYPE_V2)
+
     properties = filter.property_groups.flat
 
     if len(properties) == 0:
@@ -515,9 +558,12 @@ def _get_person_blast_radius_persons(team: Team, filter: Filter, cursor: Optiona
     select_query = _build_person_query(team, filter, return_count=False, cursor=cursor)
 
     tag_queries(product=Product.FEATURE_FLAGS, feature=Feature.QUERY)
+    # The page must be exact, so it cannot be sampled. In-order aggregation keeps its dedup
+    # memory-bounded instead, at the cost of a slower query.
     response = execute_hogql_query(
         query=select_query,
         team=team,
+        settings=bounded_memory_settings() if use_blast_radius_query_v2(team) else None,
     )
 
     # Extract person IDs from results
diff --git a/products/workflows/backend/api/hog_flow.py b/products/workflows/backend/api/hog_flow.py
index 6dcc6538155d..bb923a08514e 100644
--- a/products/workflows/backend/api/hog_flow.py
+++ b/products/workflows/backend/api/hog_flow.py
@@ -107,6 +107,7 @@
 from products.cdp.backend.models.hog_function_template import HogFunctionTemplate
 from products.cohorts.backend.models.cohort import Cohort
 from products.cohorts.backend.models.util import get_all_cohort_dependencies
+from products.feature_flags.backend.person_sampling import bounded_memory_settings
 from products.feature_flags.backend.user_blast_radius import BlastRadiusResult, get_user_blast_radius
 from products.messaging.backend.api.design_operations import apply_design_operations
 from products.messaging.backend.api.design_validation import validate_design
@@ -160,7 +161,6 @@
     parse_account_audience_filters,
 )
 from products.workflows.backend.services.audience_v2 import (
-    bounded_memory_settings,
     get_dedupe_audience_count_v2,
     get_person_audience_count_v2,
     use_audience_query_v2,
diff --git a/products/workflows/backend/services/audience_v2.py b/products/workflows/backend/services/audience_v2.py
index 4563678c595d..2ef5ae242daf 100644
--- a/products/workflows/backend/services/audience_v2.py
+++ b/products/workflows/backend/services/audience_v2.py
@@ -3,7 +3,6 @@
 import posthoganalytics
 
 from posthog.hogql import ast
-from posthog.hogql.constants import HogQLGlobalSettings
 from posthog.hogql.context import HogQLContext
 from posthog.hogql.database.database import Database
 from posthog.hogql.property import property_to_expr
@@ -13,9 +12,16 @@
 from posthog.models.filters import Filter
 from posthog.models.team.team import Team
 
+from products.feature_flags.backend.person_sampling import (
+    count_matching_persons,
+    count_settings,
+    sample_predicate,
+    sampled_or_exact_count,
+)
 from products.feature_flags.backend.user_blast_radius import (
     BlastRadiusResult,
     replace_proxy_properties,
+    sampled_person_blast_radius,
     unevaluable_filters_as_validation_errors,
 )
 from products.workflows.backend.services.batch_audience import (
@@ -26,18 +32,7 @@
 
 AUDIENCE_QUERY_V2_FLAG = "workflows-audience-query-v2"
 
-# 1-in-64 sample of the person keyspace. Sampling keys on person id (the dedup key), so every
-# version row of a sampled person lands in the sample and the argMax dedup stays exact within it.
-SAMPLE_MODULUS = 64
-
-# Below this many sampled matches the extrapolation is too noisy, so rerun exact. The relative
-# error of the estimate is about sqrt(63 / matched_persons), so 10,000 sampled matches
-# (~640k matched persons) keeps the worst case near 3%. The exact query is usually cheap in
-# that regime: few matching persons means the id prefilter keeps the dedup aggregation small.
-# Caveat: the prefilter matches any historical row version, so a churny property (many persons
-# matched once, few match now) can make the exact rerun carry a candidate set far larger than
-# the current match count suggests.
-MIN_SAMPLED_MATCHES = 10_000
+QUERY_TYPE = "workflows_audience_count_v2"
 
 
 def use_audience_query_v2(team: Team) -> bool:
@@ -50,43 +45,18 @@ def use_audience_query_v2(team: Team) -> bool:
     )
 
 
-def bounded_memory_settings() -> HogQLGlobalSettings:
-    """
-    Execution settings that keep person-dedup aggregations memory-bounded on large teams.
-
-    The person table sorts by (team_id, id), so with in-order aggregation the dedup
-    GROUP BY id streams instead of holding every person in a hash table. The spill
-    threshold is the backstop for aggregations where in-order cannot apply (for
-    example cohort subqueries): degrade to disk instead of a memory-limit error.
-    """
-    return HogQLGlobalSettings(
-        optimize_aggregation_in_order=True,
-        max_bytes_before_external_group_by=4 * 1024**3,
-    )
-
-
 def get_person_audience_count_v2(team: Team, filters: dict) -> BlastRadiusResult:
     """
     Person-audience blast radius sized from a sample of the person keyspace.
 
-    Counts run on 1 in SAMPLE_MODULUS persons and extrapolate, so cost is bounded on
-    teams where the exact dedup count exceeds the query memory limit. When a sample
-    holds too few matches for a stable extrapolation, the exact count runs instead.
+    Counts run on a sample of the persons and extrapolate, so cost is bounded on teams
+    where the exact dedup count exceeds the query memory limit.
     """
     with unevaluable_filters_as_validation_errors():
         cleaned_filter = replace_proxy_properties(team, filters)
 
         tag_queries(product=Product.WORKFLOWS, feature=Feature.QUERY)
-        # One database build shared by all counts below; each execute_hogql_query call
-        # would otherwise rebuild it, and the build cost scales with warehouse size.
-        database = Database.create_for(team=team)
-
-        total = _count_matching_persons(team, None, database)
-        if len(cleaned_filter.property_groups.flat) == 0:
-            return BlastRadiusResult(affected=total, total=total)
-
-        affected = _count_matching_persons(team, cleaned_filter, database)
-        return BlastRadiusResult(affected=min(affected, total), total=total)
+        return sampled_person_blast_radius(team, cleaned_filter, query_type=QUERY_TYPE)
 
 
 def get_dedupe_audience_count_v2(team: Team, filters: dict, dedupe_key: str) -> BlastRadiusResult:
@@ -96,7 +66,7 @@ def get_dedupe_audience_count_v2(team: Team, filters: dict, dedupe_key: str) ->
     Sampling keys on the dedupe group (the normalized email, or the person id when there is
     no email), not on the person: a group with several persons would otherwise be that many
     times more likely to land in the sample, and the estimate would drift back toward a
-    person count. Hashing the group gives every group the same 1-in-SAMPLE_MODULUS chance.
+    person count. Hashing the group gives every group the same chance.
     """
     # Defence-in-depth mirror of get_batch_audience_count: a new supported key must be
     # taught to this function too, instead of silently getting the email grouping.
@@ -109,12 +79,10 @@ def get_dedupe_audience_count_v2(team: Team, filters: dict, dedupe_key: str) ->
         tag_queries(product=Product.WORKFLOWS, feature=Feature.QUERY)
         database = Database.create_for(team=team)
 
-        total = _count_matching_persons(team, None, database)
-        sampled = _run_dedupe_count(team, cleaned_filter, database, sample_modulus=SAMPLE_MODULUS)
-        if sampled >= MIN_SAMPLED_MATCHES:
-            affected = sampled * SAMPLE_MODULUS
-        else:
-            affected = _run_dedupe_count(team, cleaned_filter, database, sample_modulus=None)
+        total = count_matching_persons(team, None, database, query_type=QUERY_TYPE)
+        affected = sampled_or_exact_count(
+            lambda sample_modulus: _run_dedupe_count(team, cleaned_filter, database, sample_modulus)
+        )
         return BlastRadiusResult(affected=min(affected, total), total=total)
 
 
@@ -123,22 +91,13 @@ def _run_dedupe_count(team: Team, filter: Filter, database: Database, sample_mod
     response = execute_hogql_query(
         query=query,
         team=team,
-        query_type="workflows_audience_count_v2",
+        query_type=QUERY_TYPE,
         context=HogQLContext(team_id=team.pk, database=database),
-        settings=_count_settings(sample_modulus),
+        settings=count_settings(sample_modulus),
     )
     return response.results[0][0] if response.results else 0
 
 
-def _count_settings(sample_modulus: Optional[int]) -> HogQLGlobalSettings:
-    # A sampled count keeps the fast parallel hash aggregation: the sample already bounds
-    # the hash table, and in-order aggregation only makes it slower. The exact runs need
-    # the streaming mode to stay memory-bounded on large teams.
-    if sample_modulus is not None:
-        return HogQLGlobalSettings(max_bytes_before_external_group_by=4 * 1024**3)
-    return bounded_memory_settings()
-
-
 def build_dedupe_count_query(team: Team, filter: Filter, sample_modulus: Optional[int]) -> ast.SelectQuery:
     where_exprs: list[ast.Expr] = [
         ast.CompareOperation(
@@ -150,19 +109,7 @@ def build_dedupe_count_query(team: Team, filter: Filter, sample_modulus: Optiona
     if sample_modulus is not None:
         # A fresh group expr per use: the resolver annotates AST nodes in place, so the
         # WHERE and SELECT must not share one instance.
-        where_exprs.append(
-            ast.CompareOperation(
-                op=ast.CompareOperationOp.Eq,
-                left=ast.Call(
-                    name="modulo",
-                    args=[
-                        ast.Call(name="cityHash64", args=[email_dedupe_group_expr()]),
-                        ast.Constant(value=sample_modulus),
-                    ],
-                ),
-                right=ast.Constant(value=0),
-            )
-        )
+        where_exprs.append(sample_predicate(email_dedupe_group_expr(), sample_modulus))
     if len(filter.property_groups.flat) > 0:
         where_exprs.append(property_to_expr(filter.property_groups, team, scope="person"))
 
@@ -171,68 +118,3 @@ def build_dedupe_count_query(team: Team, filter: Filter, sample_modulus: Optiona
         select_from=ast.JoinExpr(table=ast.Field(chain=["persons"])),
         where=ast.And(exprs=where_exprs),
     )
-
-
-def _count_matching_persons(team: Team, filter: Optional[Filter], database: Database) -> int:
-    sampled = _run_person_count(team, filter, database, sample_modulus=SAMPLE_MODULUS)
-    if sampled >= MIN_SAMPLED_MATCHES:
-        return sampled * SAMPLE_MODULUS
-    return _run_person_count(team, filter, database, sample_modulus=None)
-
-
-def _run_person_count(team: Team, filter: Optional[Filter], database: Database, sample_modulus: Optional[int]) -> int:
-    query = build_person_count_query(team, filter, sample_modulus=sample_modulus)
-    response = execute_hogql_query(
-        query=query,
-        team=team,
-        query_type="workflows_audience_count_v2",
-        context=HogQLContext(team_id=team.pk, database=database),
-        settings=_count_settings(sample_modulus),
-    )
-    return response.results[0][0] if response.results else 0
-
-
-def build_person_count_query(team: Team, filter: Optional[Filter], sample_modulus: Optional[int]) -> ast.SelectQuery:
-    where_exprs: list[ast.Expr] = [
-        ast.CompareOperation(
-            op=ast.CompareOperationOp.Eq,
-            left=ast.Field(chain=["persons", "team_id"]),
-            right=ast.Constant(value=team.pk),
-        )
-    ]
-    if sample_modulus is not None:
-        where_exprs.append(_sample_predicate(sample_modulus))
-    if filter is not None:
-        where_exprs.append(property_to_expr(filter.property_groups, team, scope="person"))
-
-    # A filter can add a one-to-many join: a `distinct_id` person property resolves through
-    # persons.pdi, which gives a person one row per distinct id. So a filtered count dedups on
-    # the person id. The unfiltered total joins nothing, so it keeps the plain count() and
-    # avoids a uniqExact state over every person on the team.
-    if filter is None:
-        count_expr: ast.Expr = ast.Call(name="count", args=[])
-    else:
-        count_expr = ast.Call(name="count", distinct=True, args=[ast.Field(chain=["persons", "id"])])
-
-    return ast.SelectQuery(
-        select=[count_expr],
-        select_from=ast.JoinExpr(table=ast.Field(chain=["persons"])),
-        where=ast.And(exprs=where_exprs),
-    )
-
-
-def _sample_predicate(modulus: int) -> ast.Expr:
-    # Built as modulo(...) calls, not the % operator: WhereClauseExtractor fails safe to
-    # "no prefilter" on ArithmeticOperation nodes, and the memory bound depends on this
-    # predicate reaching the raw person scan inside the persons lazy table.
-    return ast.CompareOperation(
-        op=ast.CompareOperationOp.Eq,
-        left=ast.Call(
-            name="modulo",
-            args=[
-                ast.Call(name="cityHash64", args=[ast.Field(chain=["persons", "id"])]),
-                ast.Constant(value=modulus),
-            ],
-        ),
-        right=ast.Constant(value=0),
-    )
diff --git a/products/workflows/backend/test/test_audience_v2.py b/products/workflows/backend/test/test_audience_v2.py
index 970a7b1258c8..943ece62344f 100644
--- a/products/workflows/backend/test/test_audience_v2.py
+++ b/products/workflows/backend/test/test_audience_v2.py
@@ -4,11 +4,10 @@
 from posthog.hogql.query import execute_hogql_query
 
 from products.cohorts.backend.models.cohort import Cohort
+from products.feature_flags.backend.person_sampling import bounded_memory_settings, build_person_count_query
 from products.feature_flags.backend.user_blast_radius import get_user_blast_radius, replace_proxy_properties
 from products.workflows.backend.services.audience_v2 import (
-    bounded_memory_settings,
     build_dedupe_count_query,
-    build_person_count_query,
     get_dedupe_audience_count_v2,
     get_person_audience_count_v2,
 )
@@ -44,8 +43,8 @@ def test_sampled_path_extrapolates_by_modulus(self):
         # Modulus 1 samples everyone and MIN_SAMPLED_MATCHES 0 forces the sampled branch,
         # so the extrapolated result must equal the exact count.
         with (
-            patch("products.workflows.backend.services.audience_v2.SAMPLE_MODULUS", 1),
-            patch("products.workflows.backend.services.audience_v2.MIN_SAMPLED_MATCHES", 0),
+            patch("products.feature_flags.backend.person_sampling.SAMPLE_MODULUS", 1),
+            patch("products.feature_flags.backend.person_sampling.MIN_SAMPLED_MATCHES", 0),
         ):
             result = get_person_audience_count_v2(self.team, FILTERS)
 
@@ -104,8 +103,8 @@ def test_distinct_id_filter_counts_a_person_with_several_aliases_once(self):
         assert (result.affected, result.total) == (v1_result.affected, v1_result.total)
 
         with (
-            patch("products.workflows.backend.services.audience_v2.SAMPLE_MODULUS", 1),
-            patch("products.workflows.backend.services.audience_v2.MIN_SAMPLED_MATCHES", 0),
+            patch("products.feature_flags.backend.person_sampling.SAMPLE_MODULUS", 1),
+            patch("products.feature_flags.backend.person_sampling.MIN_SAMPLED_MATCHES", 0),
         ):
             sampled_result = get_person_audience_count_v2(self.team, filters)
 
@@ -169,8 +168,8 @@ def test_sampled_dedupe_count_extrapolates_by_modulus(self):
         flush_persons_and_events()
 
         with (
-            patch("products.workflows.backend.services.audience_v2.SAMPLE_MODULUS", 1),
-            patch("products.workflows.backend.services.audience_v2.MIN_SAMPLED_MATCHES", 0),
+            patch("products.feature_flags.backend.person_sampling.SAMPLE_MODULUS", 1),
+            patch("products.feature_flags.backend.person_sampling.MIN_SAMPLED_MATCHES", 0),
         ):
             result = get_dedupe_audience_count_v2(self.team, FILTERS, "email")
 

From d916e6e4fb9f052c4ee30da648c8791c8c8034b6 Mon Sep 17 00:00:00 2001
From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com>
Date: Wed, 16 Sep 2026 00:03:02 +0000
Subject: [PATCH 2/7] fix(flags): keep the blast radius gate at the flags entry
 point

get_user_blast_radius is shared with workflows, which gates its own audience
counts on workflows-audience-query-v2. The flags gate sat inside that shared
function, so flags-blast-radius-query-v2 alone switched the workflows preview
and the internal node endpoint to the sampled count, outside the workflows
rollout.

Move the gate to the flag API action and give the sampled count its own
entry point, mirroring how workflows routes to its v2 count.

Generated-By: PostHog Desktop
Task-Id: b2a878ba-b8a2-442e-b16a-788f09a68e1a
---
 .../feature_flags/backend/api/feature_flag.py | 11 +++-
 .../backend/test/test_blast_radius_v2.py      | 50 ++++++++++++++++---
 .../backend/user_blast_radius.py              | 19 +++++--
 .../backend/api/test/test_hog_flow.py         | 19 +++++++
 4 files changed, 85 insertions(+), 14 deletions(-)

diff --git a/products/feature_flags/backend/api/feature_flag.py b/products/feature_flags/backend/api/feature_flag.py
index 4605260e8dd9..81296f061517 100644
--- a/products/feature_flags/backend/api/feature_flag.py
+++ b/products/feature_flags/backend/api/feature_flag.py
@@ -130,7 +130,11 @@
     teams_linking_flag_in_project,
 )
 from products.feature_flags.backend.types import PropertyFilterType
-from products.feature_flags.backend.user_blast_radius import get_user_blast_radius
+from products.feature_flags.backend.user_blast_radius import (
+    get_person_blast_radius_v2,
+    get_user_blast_radius,
+    use_blast_radius_query_v2,
+)
 from products.feature_flags.backend.version_history import (
     VersionHistoryIncomplete,
     VersionNotFound,
@@ -4665,7 +4669,10 @@ def user_blast_radius(self, request: request.Request, **kwargs):
         condition = request.data.get("condition") or {}
         group_type_index = request.data.get("group_type_index", None)
 
-        result = get_user_blast_radius(self.team, condition, group_type_index)
+        if group_type_index is None and use_blast_radius_query_v2(self.team):
+            result = get_person_blast_radius_v2(self.team, condition)
+        else:
+            result = get_user_blast_radius(self.team, condition, group_type_index)
 
         return Response({"affected": result.affected, "total": result.total})
 
diff --git a/products/feature_flags/backend/test/test_blast_radius_v2.py b/products/feature_flags/backend/test/test_blast_radius_v2.py
index 0fc1723b3987..7bb923f9fdb5 100644
--- a/products/feature_flags/backend/test/test_blast_radius_v2.py
+++ b/products/feature_flags/backend/test/test_blast_radius_v2.py
@@ -1,16 +1,20 @@
-from posthog.test.base import BaseTest, ClickhouseTestMixin, _create_person, flush_persons_and_events
+from posthog.test.base import APIBaseTest, ClickhouseTestMixin, _create_person, flush_persons_and_events
 from unittest.mock import patch
 
 from parameterized import parameterized
 
-from products.feature_flags.backend.user_blast_radius import get_user_blast_radius
+from products.feature_flags.backend.user_blast_radius import (
+    BlastRadiusResult,
+    get_person_blast_radius_v2,
+    get_user_blast_radius,
+)
 
 FILTERS = {"properties": [{"key": "subscribed", "type": "person", "value": ["true"], "operator": "exact"}]}
 
-GATE = "products.feature_flags.backend.user_blast_radius.use_blast_radius_query_v2"
+GATE = "products.feature_flags.backend.api.feature_flag.use_blast_radius_query_v2"
 
 
-class TestBlastRadiusQueryV2(ClickhouseTestMixin, BaseTest):
+class TestBlastRadiusQueryV2(ClickhouseTestMixin, APIBaseTest):
     def _create_persons(self, subscribed_flags: list[bool]) -> None:
         for i, subscribed in enumerate(subscribed_flags, start=1):
             _create_person(
@@ -32,8 +36,7 @@ def test_sized_condition_matches_the_v1_counts(self, _name, filters, affected, t
         self._create_persons([True, True, True, False])
 
         v1_result = get_user_blast_radius(self.team, filters)
-        with patch(GATE, return_value=True):
-            result = get_user_blast_radius(self.team, filters)
+        result = get_person_blast_radius_v2(self.team, filters)
 
         assert (result.affected, result.total) == (affected, total)
         assert (result.affected, result.total) == (v1_result.affected, v1_result.total)
@@ -44,10 +47,41 @@ def test_sampled_path_extrapolates_by_modulus(self):
         # Modulus 1 samples everyone and MIN_SAMPLED_MATCHES 0 forces the sampled branch,
         # so the extrapolated result must equal the exact count.
         with (
-            patch(GATE, return_value=True),
             patch("products.feature_flags.backend.person_sampling.SAMPLE_MODULUS", 1),
             patch("products.feature_flags.backend.person_sampling.MIN_SAMPLED_MATCHES", 0),
         ):
-            result = get_user_blast_radius(self.team, FILTERS)
+            result = get_person_blast_radius_v2(self.team, FILTERS)
 
         assert (result.affected, result.total) == (2, 3)
+
+    @parameterized.expand(
+        [
+            ("person condition, gate on", True, None, True),
+            ("person condition, gate off", False, None, False),
+            # The sampled count only covers person audiences, so a group condition stays on v1.
+            ("group condition, gate on", True, 0, False),
+        ]
+    )
+    def test_endpoint_uses_the_sampled_count_only_for_a_gated_person_condition(
+        self, _name, gate_on, group_type_index, expects_v2
+    ):
+        body: dict = {"condition": {"properties": []}}
+        if group_type_index is not None:
+            body["group_type_index"] = group_type_index
+
+        with (
+            patch(GATE, return_value=gate_on),
+            patch("products.feature_flags.backend.api.feature_flag.get_person_blast_radius_v2") as sampled,
+            patch("products.feature_flags.backend.api.feature_flag.get_user_blast_radius") as v1,
+        ):
+            sampled.return_value = BlastRadiusResult(affected=6400, total=64000)
+            v1.return_value = BlastRadiusResult(affected=4, total=10)
+
+            response = self.client.post(
+                f"/api/projects/{self.team.id}/feature_flags/user_blast_radius",
+                body,
+            )
+
+        assert response.status_code == 200, response.json()
+        assert response.json()["affected"] == (6400 if expects_v2 else 4)
+        assert sampled.called is expects_v2
diff --git a/products/feature_flags/backend/user_blast_radius.py b/products/feature_flags/backend/user_blast_radius.py
index 9c73ca9dcc6f..a5e00d356af8 100644
--- a/products/feature_flags/backend/user_blast_radius.py
+++ b/products/feature_flags/backend/user_blast_radius.py
@@ -169,6 +169,21 @@ def get_user_blast_radius(
             return _get_person_blast_radius(team, cleaned_filter)
 
 
+def get_person_blast_radius_v2(team: Team, feature_flag_condition: dict) -> BlastRadiusResult:
+    """
+    Flags-owned entry point for the sampled person count, behind flags-blast-radius-query-v2.
+
+    The gate is applied by the caller, not inside get_user_blast_radius: workflows shares that
+    function and gates its own audience counts on workflows-audience-query-v2, so a gate in
+    there would move workflows counts outside the workflows rollout.
+    """
+    with unevaluable_filters_as_validation_errors():
+        cleaned_filter = replace_proxy_properties(team, feature_flag_condition)
+
+        tag_queries(product=Product.FEATURE_FLAGS, feature=Feature.QUERY)
+        return sampled_person_blast_radius(team, cleaned_filter, query_type=QUERY_TYPE_V2)
+
+
 def get_user_blast_radius_persons(
     team: Team,
     feature_flag_condition: dict,
@@ -188,10 +203,6 @@ def get_user_blast_radius_persons(
 def _get_person_blast_radius(team: Team, filter: Filter) -> BlastRadiusResult:
     """Calculate blast radius for person-based feature flags using HogQL."""
 
-    if use_blast_radius_query_v2(team):
-        tag_queries(product=Product.FEATURE_FLAGS, feature=Feature.QUERY)
-        return sampled_person_blast_radius(team, filter, query_type=QUERY_TYPE_V2)
-
     properties = filter.property_groups.flat
 
     if len(properties) == 0:
diff --git a/products/workflows/backend/api/test/test_hog_flow.py b/products/workflows/backend/api/test/test_hog_flow.py
index b96b58f22612..d247781709fd 100644
--- a/products/workflows/backend/api/test/test_hog_flow.py
+++ b/products/workflows/backend/api/test/test_hog_flow.py
@@ -3230,6 +3230,25 @@ def test_hog_flow_user_blast_radius_routes_to_v2_when_flag_enabled(self):
             mock_v1.assert_called_once()
             mock_v2.assert_called_once()
 
+    def test_hog_flow_user_blast_radius_ignores_the_feature_flags_gate(self):
+        # Workflows counts follow workflows-audience-query-v2 only. The flags product gates its
+        # own sampled count on a separate flag, and that gate must not reach this endpoint: a
+        # sampled count here would move workflows numbers outside the workflows rollout.
+        # The routing test above mocks get_user_blast_radius away, so it cannot see this.
+        with (
+            patch("products.workflows.backend.api.hog_flow.use_audience_query_v2", return_value=False),
+            patch("products.feature_flags.backend.user_blast_radius.use_blast_radius_query_v2", return_value=True),
+            patch("products.feature_flags.backend.user_blast_radius.sampled_person_blast_radius") as mock_sampled,
+        ):
+            response = self.client.post(
+                f"/api/projects/{self.team.id}/hog_flows/user_blast_radius",
+                {"filters": {"properties": []}},
+            )
+
+        assert response.status_code == 200, response.json()
+        assert response.json()["total"] == self.team.persons_seen_so_far
+        mock_sampled.assert_not_called()
+
     @override_settings(
         HOGFLOW_BATCH_TRIGGER_LIMIT=5000,
         HOGFLOW_BATCH_TRIGGER_LIMIT_ELEVATED=50000,

From 00184043c6e09e54fc1f6c9bf4872dbf3c82310f Mon Sep 17 00:00:00 2001
From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com>
Date: Wed, 16 Sep 2026 00:09:27 +0000
Subject: [PATCH 3/7] fix(flags): throw instead of returning a partial person
 count

timeout_overflow_mode was left unset, so both settings constructors inherited
the cluster profile's overflow behavior. Under a "break" profile a timed-out
aggregate returns whatever it read so far, and the sampled count then
multiplies that partial tally by the modulus and reports it as an audience
size. The exact person page has the same problem: a short page is a send that
silently skips recipients.

Set timeout_overflow_mode="throw" in both constructors, as the setting's
contract in posthog/hogql/constants.py prescribes for sampling queries whose
output feeds further computation.

Generated-By: PostHog Desktop
Task-Id: b2a878ba-b8a2-442e-b16a-788f09a68e1a
---
 .../feature_flags/backend/person_sampling.py  | 12 +++++++++--
 .../backend/test/test_person_sampling.py      | 20 +++++++++++++++++++
 2 files changed, 30 insertions(+), 2 deletions(-)
 create mode 100644 products/feature_flags/backend/test/test_person_sampling.py

diff --git a/products/feature_flags/backend/person_sampling.py b/products/feature_flags/backend/person_sampling.py
index fa0f4b290e40..87ea5df453f3 100644
--- a/products/feature_flags/backend/person_sampling.py
+++ b/products/feature_flags/backend/person_sampling.py
@@ -38,6 +38,10 @@ def bounded_memory_settings() -> HogQLGlobalSettings:
     example cohort subqueries): degrade to disk instead of a memory-limit error.
     """
     return HogQLGlobalSettings(
+        # A partial result is worse than an error here: the counts get extrapolated and the
+        # person pages get sent to, so a timed-out query must fail rather than come up short.
+        # Left unset it inherits the cluster profile, which may accept partial results.
+        timeout_overflow_mode="throw",
         optimize_aggregation_in_order=True,
         max_bytes_before_external_group_by=4 * 1024**3,
     )
@@ -46,9 +50,13 @@ def bounded_memory_settings() -> HogQLGlobalSettings:
 def count_settings(sample_modulus: Optional[int]) -> HogQLGlobalSettings:
     # A sampled count keeps the fast parallel hash aggregation: the sample already bounds
     # the hash table, and in-order aggregation only makes it slower. The exact runs need
-    # the streaming mode to stay memory-bounded on large teams.
+    # the streaming mode to stay memory-bounded on large teams. Both throw on timeout,
+    # because the sampled count multiplies whatever it read by the modulus.
     if sample_modulus is not None:
-        return HogQLGlobalSettings(max_bytes_before_external_group_by=4 * 1024**3)
+        return HogQLGlobalSettings(
+            timeout_overflow_mode="throw",
+            max_bytes_before_external_group_by=4 * 1024**3,
+        )
     return bounded_memory_settings()
 
 
diff --git a/products/feature_flags/backend/test/test_person_sampling.py b/products/feature_flags/backend/test/test_person_sampling.py
new file mode 100644
index 000000000000..de90d0534674
--- /dev/null
+++ b/products/feature_flags/backend/test/test_person_sampling.py
@@ -0,0 +1,20 @@
+from django.test import SimpleTestCase
+
+from parameterized import parameterized
+
+from products.feature_flags.backend.person_sampling import bounded_memory_settings, count_settings
+
+
+class TestPersonSamplingSettings(SimpleTestCase):
+    @parameterized.expand(
+        [
+            ("memory bounded", bounded_memory_settings),
+            ("sampled count", lambda: count_settings(64)),
+            ("exact count", lambda: count_settings(None)),
+        ]
+    )
+    def test_settings_reject_a_partial_result_on_timeout(self, _name, build_settings):
+        # Left unset, the mode inherits the cluster profile, which may return whatever the
+        # query aggregated before the timeout. The sampled count would then multiply that
+        # partial tally by the modulus and report it as an audience size.
+        assert build_settings().timeout_overflow_mode == "throw"

From 9329a40809e3d820d716c15ca181e3ed0eb86b4c Mon Sep 17 00:00:00 2001
From: Phil Haack <haacked@gmail.com>
Date: Tue, 15 Sep 2026 17:34:06 -0700
Subject: [PATCH 4/7] chore(flags): drop a gate read from an unreachable query
 path

_get_person_blast_radius_persons has no production caller: the only route into
get_user_blast_radius_persons is the workflows batch audience, which delegates
just its group branch, and that lands in the group query. Passing the bounded
settings there read a flags feature flag inside a function workflows shares,
for a page nothing requests.

Restore the call to its previous form.

Generated-By: PostHog Desktop
Task-Id: be8e214b-4146-4efa-8f3c-5bd3bd59210a
---
 products/feature_flags/backend/user_blast_radius.py | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/products/feature_flags/backend/user_blast_radius.py b/products/feature_flags/backend/user_blast_radius.py
index a5e00d356af8..ee0a450e41c1 100644
--- a/products/feature_flags/backend/user_blast_radius.py
+++ b/products/feature_flags/backend/user_blast_radius.py
@@ -29,7 +29,7 @@
 from posthog.models.team.team import Team
 
 from products.cohorts.backend.models.cohort import Cohort
-from products.feature_flags.backend.person_sampling import bounded_memory_settings, count_matching_persons
+from products.feature_flags.backend.person_sampling import count_matching_persons
 
 
 @frozen
@@ -569,12 +569,9 @@ def _get_person_blast_radius_persons(team: Team, filter: Filter, cursor: Optiona
     select_query = _build_person_query(team, filter, return_count=False, cursor=cursor)
 
     tag_queries(product=Product.FEATURE_FLAGS, feature=Feature.QUERY)
-    # The page must be exact, so it cannot be sampled. In-order aggregation keeps its dedup
-    # memory-bounded instead, at the cost of a slower query.
     response = execute_hogql_query(
         query=select_query,
         team=team,
-        settings=bounded_memory_settings() if use_blast_radius_query_v2(team) else None,
     )
 
     # Extract person IDs from results

From 7264a8e8a0725365d233f530bb94ef7e2ce6a04d Mon Sep 17 00:00:00 2001
From: Phil Haack <haacked@gmail.com>
Date: Tue, 15 Sep 2026 18:17:39 -0700
Subject: [PATCH 5/7] fix(flags): evaluate the blast radius gate locally

The release-conditions editor calls the sizing endpoint once per condition group,
so a gate that can fall back to a network fetch puts a flag round trip in front of
an interactive request.

Read the flag locally only, and target the project group rather than a synthetic
distinct id: local evaluation can resolve a group property carried on the call,
where a person property would need the fetch it is meant to avoid. Same shape the
recent-activity sizing gate uses.

Generated-By: PostHog Desktop
Task-Id: be8e214b-4146-4efa-8f3c-5bd3bd59210a
---
 .../feature_flags/backend/user_blast_radius.py | 18 +++++++++++-------
 1 file changed, 11 insertions(+), 7 deletions(-)

diff --git a/products/feature_flags/backend/user_blast_radius.py b/products/feature_flags/backend/user_blast_radius.py
index ee0a450e41c1..e6c5de7d4447 100644
--- a/products/feature_flags/backend/user_blast_radius.py
+++ b/products/feature_flags/backend/user_blast_radius.py
@@ -4,7 +4,6 @@
 
 from django.core.exceptions import ObjectDoesNotExist
 
-import posthoganalytics
 from rest_framework.exceptions import ValidationError
 
 from posthog.schema import PropertyOperator
@@ -27,6 +26,7 @@
 from posthog.models.property import GroupTypeIndex, Property, PropertyGroup, PropertyValidationError
 from posthog.models.property.relative_date import relative_date_parse_for_feature_flag_matching
 from posthog.models.team.team import Team
+from posthog.ph_client import feature_enabled_or_false
 
 from products.cohorts.backend.models.cohort import Cohort
 from products.feature_flags.backend.person_sampling import count_matching_persons
@@ -44,12 +44,16 @@ class BlastRadiusResult:
 
 
 def use_blast_radius_query_v2(team: Team) -> bool:
-    return bool(
-        posthoganalytics.feature_enabled(
-            BLAST_RADIUS_QUERY_V2_FLAG,
-            str(team.uuid),
-            send_feature_flag_events=False,
-        )
+    # Local-only, so a sizing request never waits on a flag fetch. That is also why the gate
+    # targets the project group: the project id travels with the call, where a person property
+    # would need the fetch to answer.
+    return feature_enabled_or_false(
+        BLAST_RADIUS_QUERY_V2_FLAG,
+        f"team-{team.pk}",
+        groups={"project": str(team.pk)},
+        group_properties={"project": {"id": str(team.pk)}},
+        only_evaluate_locally=True,
+        send_feature_flag_events=False,
     )
 
 

From 188c8e2ffa6f009d5f98f7432bb7728f512c9dc7 Mon Sep 17 00:00:00 2001
From: Phil Haack <haacked@gmail.com>
Date: Wed, 16 Sep 2026 10:09:09 -0700
Subject: [PATCH 6/7] chore(flags): say when the v1 argmax pin is load-bearing

The comment read as though a v2-pinned team would cap this count near 101 rows.
It cannot: the team_id condition sends the query down the persons table's
prefilter path, which emits the v1 argmax dedup whatever the modifier says.

Keep the pin, since losing that path would silently under-report the
denominator, but say that it only matters in that case so a reader does not
take it for an active guard.

Generated-By: PostHog Desktop
Task-Id: be8e214b-4146-4efa-8f3c-5bd3bd59210a
---
 products/feature_flags/backend/person_sampling.py | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/products/feature_flags/backend/person_sampling.py b/products/feature_flags/backend/person_sampling.py
index 87ea5df453f3..5b00009ac40a 100644
--- a/products/feature_flags/backend/person_sampling.py
+++ b/products/feature_flags/backend/person_sampling.py
@@ -90,8 +90,11 @@ def _run_person_count(
         query_type=query_type,
         # Pin v1 for the unfiltered count, as count_persons_seen_so_far does: the v2 persons
         # path pushes the executor's default LIMIT into its dedup subquery, which caps a bare
-        # count() at ~101 for teams pinned to v2 (see #87323). A filtered count reads person
-        # properties, where v2 is the faster shape, so it keeps the automatic choice.
+        # count() at ~101 for teams pinned to v2 (see #87323). The team_id condition already
+        # sends these queries down the persons table's prefilter path, which emits the v1 dedup
+        # whatever this says, so the pin is only load-bearing if that path is ever lost. A
+        # filtered count reads person properties, where v2 is the faster shape, so it keeps the
+        # automatic choice.
         modifiers=HogQLQueryModifiers(personsArgMaxVersion=PersonsArgMaxVersion.V1) if filter is None else None,
         context=HogQLContext(team_id=team.pk, database=database),
         settings=count_settings(sample_modulus),

From a780bfe5a57369194c0c77bf7806a77965152d86 Mon Sep 17 00:00:00 2001
From: Phil Haack <haacked@gmail.com>
Date: Wed, 16 Sep 2026 12:50:44 -0700
Subject: [PATCH 7/7] test(flags): pin the sampling threshold, the gate kwargs,
 and the enumeration timeout

The extrapolation test drove the count with SAMPLE_MODULUS patched to 1, where the
multiply is a no-op, so dropping the extrapolation entirely still passed. Drive
sampled_or_exact_count with a stub instead, which pins the multiplier and both sides
of the threshold boundary without a database.

Add the two cases nothing covered. The gate's only_evaluate_locally and
send_feature_flag_events kwargs now have a test, following the one the cache gates
already carry: dropping either is silent, and dropping the group ids makes local
evaluation read False for every team. The workflows enumeration path now has one
too, because throw-on-timeout changed its failure mode from a short page that ends
pagination quietly to a raise.

Also correct two comments. The blast radius docstring said HogQL cannot spill to
disk while the settings it uses set a spill threshold; only the defaults do not
spill. The threshold comment called the exact rerun cheap without saying the scan
is still full-table, which reads as a full-table dedup.

Generated-By: PostHog Desktop
Task-Id: be8e214b-4146-4efa-8f3c-5bd3bd59210a
---
 .../feature_flags/backend/person_sampling.py  |  9 +++-
 .../backend/test/test_blast_radius_v2.py      | 25 ++++++++++
 .../backend/test/test_person_sampling.py      | 46 ++++++++++++++++++-
 .../backend/user_blast_radius.py              | 10 ++--
 .../backend/api/test/test_hog_flow.py         | 25 +++++++---
 5 files changed, 102 insertions(+), 13 deletions(-)

diff --git a/products/feature_flags/backend/person_sampling.py b/products/feature_flags/backend/person_sampling.py
index 5b00009ac40a..2676bf8c98d7 100644
--- a/products/feature_flags/backend/person_sampling.py
+++ b/products/feature_flags/backend/person_sampling.py
@@ -20,8 +20,13 @@
 
 # Below this many sampled matches the extrapolation is too noisy, so rerun exact. The relative
 # error of the estimate is about sqrt(63 / matched_persons), so 10,000 sampled matches
-# (~640k matched persons) keeps the worst case near 3%. The exact query is usually cheap in
-# that regime: few matching persons means the id prefilter keeps the dedup aggregation small.
+# (~640k matched persons) keeps the worst case near 3%.
+#
+# The threshold reads the match count, not the table size, so a large team whose condition
+# matches few persons reruns exact. That rerun reads the whole table, but it does not rebuild
+# the shape this sampling exists to avoid: the id prefilter holds the dedup to the matched
+# candidate set, and count_settings adds in-order aggregation and a spill threshold. The cost
+# is scan time instead of a memory-limit error.
 # Caveat: the prefilter matches any historical row version, so a churny property (many persons
 # matched once, few match now) can make the exact rerun carry a candidate set far larger than
 # the current match count suggests.
diff --git a/products/feature_flags/backend/test/test_blast_radius_v2.py b/products/feature_flags/backend/test/test_blast_radius_v2.py
index 7bb923f9fdb5..a80826197ce6 100644
--- a/products/feature_flags/backend/test/test_blast_radius_v2.py
+++ b/products/feature_flags/backend/test/test_blast_radius_v2.py
@@ -1,12 +1,17 @@
 from posthog.test.base import APIBaseTest, ClickhouseTestMixin, _create_person, flush_persons_and_events
 from unittest.mock import patch
 
+from django.test import SimpleTestCase
+
 from parameterized import parameterized
 
+from posthog.models.team.team import Team
+
 from products.feature_flags.backend.user_blast_radius import (
     BlastRadiusResult,
     get_person_blast_radius_v2,
     get_user_blast_radius,
+    use_blast_radius_query_v2,
 )
 
 FILTERS = {"properties": [{"key": "subscribed", "type": "person", "value": ["true"], "operator": "exact"}]}
@@ -85,3 +90,23 @@ def test_endpoint_uses_the_sampled_count_only_for_a_gated_person_condition(
         assert response.status_code == 200, response.json()
         assert response.json()["affected"] == (6400 if expects_v2 else 4)
         assert sampled.called is expects_v2
+
+
+class TestBlastRadiusGate(SimpleTestCase):
+    def test_the_gate_evaluates_locally_and_captures_nothing(self):
+        with patch(
+            "products.feature_flags.backend.user_blast_radius.feature_enabled_or_false", return_value=True
+        ) as feature_enabled:
+            assert use_blast_radius_query_v2(Team(pk=42)) is True
+
+        kwargs = feature_enabled.call_args.kwargs
+        # A remote evaluation would put a flags fetch in front of the sizing request the
+        # release-conditions editor sends per condition group, and an event capture would
+        # bill each of those requests as product usage.
+        assert kwargs["only_evaluate_locally"] is True
+        assert kwargs["send_feature_flag_events"] is False
+        # The flag targets the project group, and local evaluation can only resolve that
+        # condition from the ids carried on the call. Without them the gate reads False
+        # for every team, whatever the rollout says.
+        assert kwargs["groups"] == {"project": "42"}
+        assert kwargs["group_properties"] == {"project": {"id": "42"}}
diff --git a/products/feature_flags/backend/test/test_person_sampling.py b/products/feature_flags/backend/test/test_person_sampling.py
index de90d0534674..cd2a6a3e74cb 100644
--- a/products/feature_flags/backend/test/test_person_sampling.py
+++ b/products/feature_flags/backend/test/test_person_sampling.py
@@ -1,8 +1,17 @@
+from collections.abc import Callable
+from typing import Optional
+
 from django.test import SimpleTestCase
 
 from parameterized import parameterized
 
-from products.feature_flags.backend.person_sampling import bounded_memory_settings, count_settings
+from products.feature_flags.backend.person_sampling import (
+    MIN_SAMPLED_MATCHES,
+    SAMPLE_MODULUS,
+    bounded_memory_settings,
+    count_settings,
+    sampled_or_exact_count,
+)
 
 
 class TestPersonSamplingSettings(SimpleTestCase):
@@ -18,3 +27,38 @@ def test_settings_reject_a_partial_result_on_timeout(self, _name, build_settings
         # query aggregated before the timeout. The sampled count would then multiply that
         # partial tally by the modulus and report it as an audience size.
         assert build_settings().timeout_overflow_mode == "throw"
+
+
+class TestSampledOrExactCount(SimpleTestCase):
+    # Distinct from every extrapolation below, so the returned value names the branch that ran.
+    EXACT_COUNT = 7
+
+    def _stub_count(self, sampled: int) -> tuple[Callable[[Optional[int]], int], list[Optional[int]]]:
+        moduli: list[Optional[int]] = []
+
+        def run_count(sample_modulus: Optional[int]) -> int:
+            moduli.append(sample_modulus)
+            return sampled if sample_modulus is not None else self.EXACT_COUNT
+
+        return run_count, moduli
+
+    @parameterized.expand(
+        [
+            ("above the threshold", MIN_SAMPLED_MATCHES + 1),
+            # The boundary itself: `>` in place of `>=` reruns exact here and fails.
+            ("at the threshold", MIN_SAMPLED_MATCHES),
+        ]
+    )
+    def test_a_large_enough_sample_is_extrapolated_without_an_exact_rerun(self, _name, sampled):
+        run_count, moduli = self._stub_count(sampled)
+
+        # Returning the sample unmultiplied would under-report the audience by the modulus.
+        assert sampled_or_exact_count(run_count) == sampled * SAMPLE_MODULUS
+        assert moduli == [SAMPLE_MODULUS]
+
+    def test_a_small_sample_is_discarded_for_an_exact_count(self):
+        run_count, moduli = self._stub_count(MIN_SAMPLED_MATCHES - 1)
+
+        assert sampled_or_exact_count(run_count) == self.EXACT_COUNT
+        # The probe still runs and is paid for before the exact count replaces it.
+        assert moduli == [SAMPLE_MODULUS, None]
diff --git a/products/feature_flags/backend/user_blast_radius.py b/products/feature_flags/backend/user_blast_radius.py
index e6c5de7d4447..fc1682af861a 100644
--- a/products/feature_flags/backend/user_blast_radius.py
+++ b/products/feature_flags/backend/user_blast_radius.py
@@ -62,10 +62,12 @@ def sampled_person_blast_radius(team: Team, filter: Filter, query_type: str) ->
     Person blast radius whose peak query memory does not grow with the size of the person table.
 
     The exact counts dedup the person table with a hash GROUP BY that holds every matched
-    person in memory, and HogQL queries cannot spill to disk, so on a large team both counts
-    cross the per-query memory limit and the caller gets an error instead of a number. These
-    counts read a sample of the persons and extrapolate, and fall back to an exact, streaming
-    count when the sample holds too few matches to extrapolate from.
+    person in memory, and HogQL's default settings set no spill threshold, so on a large team
+    both counts cross the per-query memory limit and the caller gets an error instead of a
+    number. These counts read a sample of the persons and extrapolate, and fall back to an
+    exact count when the sample holds too few matches to extrapolate from. That fallback stays
+    bounded because count_settings overrides those defaults with in-order aggregation and a
+    spill threshold.
     """
     # One database build shared by both counts; each execute_hogql_query call would otherwise
     # rebuild it, and the build cost scales with the team's warehouse size.
diff --git a/products/workflows/backend/api/test/test_hog_flow.py b/products/workflows/backend/api/test/test_hog_flow.py
index d247781709fd..8833a7147a0d 100644
--- a/products/workflows/backend/api/test/test_hog_flow.py
+++ b/products/workflows/backend/api/test/test_hog_flow.py
@@ -6,7 +6,7 @@
 from typing import Any, Optional
 
 from posthog.test.base import APIBaseTest, ClickhouseTestMixin
-from unittest.mock import MagicMock, PropertyMock, patch
+from unittest.mock import ANY, MagicMock, PropertyMock, patch
 
 from django.core.management import call_command
 from django.db import connection
@@ -3380,11 +3380,20 @@ def test_internal_user_blast_radius_persons_rejects_flag_condition(self):
         assert "Feature flags can't be used as a batch audience condition" in response.json().get("error", "")
         mock_get_batch_audience_person_ids.assert_not_called()
 
+    @parameterized.expand(
+        [
+            ("gate off", False, None),
+            ("gate on", True, "throw"),
+        ]
+    )
     @override_settings(INTERNAL_API_SECRET="test-secret-123")
-    def test_internal_user_blast_radius_persons_uses_workflows_query(self):
-        with patch(
-            "products.workflows.backend.api.hog_flow.get_batch_audience_person_ids", return_value=["id-1"]
-        ) as mock_workflows_query:
+    def test_internal_user_blast_radius_persons_uses_workflows_query(self, _name, gate_on, expected_timeout_mode):
+        with (
+            patch("products.workflows.backend.api.hog_flow.use_audience_query_v2", return_value=gate_on),
+            patch(
+                "products.workflows.backend.api.hog_flow.get_batch_audience_person_ids", return_value=["id-1"]
+            ) as mock_workflows_query,
+        ):
             response = self.client.post(
                 f"/api/projects/{self.team.id}/internal/hog_flows/user_blast_radius_persons",
                 {"filters": {"properties": []}, "dedupe_key": "email"},
@@ -3395,8 +3404,12 @@ def test_internal_user_blast_radius_persons_uses_workflows_query(self):
         assert response.status_code == 200, response.json()
         assert response.json()["users_affected"] == ["id-1"]
         mock_workflows_query.assert_called_once_with(
-            self.team, {"properties": []}, None, None, dedupe_key="email", settings=None
+            self.team, {"properties": []}, None, None, dedupe_key="email", settings=ANY
         )
+        # Gated on, a timed-out page has to raise instead of coming back short. A short page
+        # reads as the end of the audience, so the batch send skips every recipient after it.
+        passed_settings = mock_workflows_query.call_args.kwargs["settings"]
+        assert getattr(passed_settings, "timeout_overflow_mode", None) == expected_timeout_mode
 
     @parameterized.expand(
         [