feat(insights): launch funnels as a Clickhouse UDF behind a feature flag

docker-compose.dev-full.yml

@@ -47,6 +47,8 @@ services:
             - ./docker/clickhouse/docker-entrypoint-initdb.d:/docker-entrypoint-initdb.d
             - ./docker/clickhouse/config.xml:/etc/clickhouse-server/config.xml
             - ./docker/clickhouse/users-dev.xml:/etc/clickhouse-server/users.xml
+            - ./docker/clickhouse/test_function.xml:/etc/clickhouse-server/test_function.xml
+            - ./posthog/users_scripts:/var/lib/clickhouse/user_scripts
         depends_on:
             - kafka
             - zookeeper

docker-compose.dev.yml

@@ -65,6 +65,8 @@ services:
             - ./docker/clickhouse/docker-entrypoint-initdb.d:/docker-entrypoint-initdb.d
             - ./docker/clickhouse/config.xml:/etc/clickhouse-server/config.xml
             - ./docker/clickhouse/users-dev.xml:/etc/clickhouse-server/users.xml
+            - ./docker/clickhouse/test_function.xml:/etc/clickhouse-server/test_function.xml
+            - ./posthog/users_scripts:/var/lib/clickhouse/user_scripts
         extra_hosts:
             - 'host.docker.internal:host-gateway'
         depends_on:

docker/clickhouse/docker-entrypoint-initdb.d/init-db.sh

@@ -1,4 +1,5 @@
 #!/bin/bash
 set -e
 
+apk add python3
 cp -r /idl/* /var/lib/clickhouse/format_schemas/

docker/clickhouse/test_function.xml

@@ -0,0 +1,129 @@
+<functions>
+    <function>
+        <type>executable</type>
+        <name>aggregate_funnel</name>
+        <return_type>Array(Tuple(Int8, Nullable(String), Array(Float64)))</return_type>
+        <argument>
+            <type>UInt8</type>
+            <name>num_steps</name>
+        </argument>
+        <argument>
+            <type>UInt64</type>
+            <name>conversion_window_limit</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>breakdown_attribution_type</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>funnel_order_type</name>
+        </argument>
+        <argument>
+            <type>Array(Nullable(String))</type>
+            <name>prop_vals</name>
+        </argument>
+        <argument>
+            <type>Array(Tuple(Nullable(Float64), Nullable(String), Array(Int8)))</type>
+            <name>value</name>
+        </argument>
+        <format>TabSeparated</format>
+        <command>aggregate_funnel.py</command>
+    </function>
+
+    <function>
+        <type>executable</type>
+        <name>aggregate_funnel_cohort</name>
+        <return_type>Array(Tuple(Int8, UInt64, Array(Float64)))</return_type>
+        <argument>
+            <type>UInt8</type>
+            <name>num_steps</name>
+        </argument>
+        <argument>
+            <type>UInt64</type>
+            <name>conversion_window_limit</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>breakdown_attribution_type</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>funnel_order_type</name>
+        </argument>
+        <argument>
+            <type>Array(UInt64)</type>
+            <name>prop_vals</name>
+        </argument>
+        <argument>
+            <type>Array(Tuple(Nullable(Float64), UInt64, Array(Int8)))</type>
+            <name>value</name>
+        </argument>
+        <format>TabSeparated</format>
+        <command>aggregate_funnel_cohort.py</command>
+    </function>
+
+    <function>
+        <type>executable</type>
+        <name>aggregate_funnel_array</name>
+        <return_type>Array(Tuple(Int8, Array(String), Array(Float64)))</return_type>
+        <argument>
+            <type>UInt8</type>
+            <name>num_steps</name>
+        </argument>
+        <argument>
+            <type>UInt64</type>
+            <name>conversion_window_limit</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>breakdown_attribution_type</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>funnel_order_type</name>
+        </argument>
+        <argument>
+            <type>Array(Array(String))</type>
+            <name>prop_vals</name>
+        </argument>
+        <argument>
+            <type>Array(Tuple(Nullable(Float64), Array(String), Array(Int8)))</type>
+            <name>value</name>
+        </argument>
+        <format>TabSeparated</format>
+        <command>aggregate_funnel_array.py</command>
+    </function>
+
+    <function>
+        <type>executable</type>
+        <name>aggregate_funnel_test</name>
+        <return_type>String</return_type>
+        <argument>
+            <type>UInt8</type>
+            <name>num_steps</name>
+        </argument>
+        <argument>
+            <type>UInt64</type>
+            <name>conversion_window_limit</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>breakdown_attribution_type</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>funnel_order_type</name>
+        </argument>
+        <argument>
+            <type>Array(Array(String))</type>
+            <name>prop_vals</name>
+        </argument>
+        <argument>
+            <type>Array(Tuple(Nullable(Float64), Nullable(String), Array(Int8)))</type>
+            <name>value</name>
+        </argument>
+        <format>TabSeparated</format>
+        <command>test_function.py</command>
+    </function>
+</functions>

frontend/src/lib/constants.tsx

@@ -206,6 +206,7 @@ export const FEATURE_FLAGS = {
     MULTIPLE_BREAKDOWNS: 'multiple-breakdowns', // owner: @skoob13 #team-product-analytics
     WEB_ANALYTICS_LIVE_USER_COUNT: 'web-analytics-live-user-count', // owner: @robbie-c
     SETTINGS_SESSION_TABLE_VERSION: 'settings-session-table-version', // owner: @robbie-c
+    INSIGHT_FUNNELS_USE_UDF: 'insight-funnels-use-udf', // owner: @aspicer #team-product-analytics
     FIRST_TIME_FOR_USER_MATH: 'first-time-for-user-math', // owner: @skoob13 #team-product-analytics
     MULTITAB_EDITOR: 'multitab-editor', // owner: @EDsCODE #team-data-warehouse
     WEB_ANALYTICS_REPLAY: 'web-analytics-replay', // owner: @robbie-c

posthog/hogql/functions/mapping.py

@@ -470,6 +470,7 @@ def compare_types(arg_types: list[ConstantType], sig_arg_types: tuple[ConstantTy
     "timeStampSub": HogQLFunctionMeta("timeStampSub", 2, 2),
     "now": HogQLFunctionMeta("now64", 0, 1, tz_aware=True, case_sensitive=False),
     "nowInBlock": HogQLFunctionMeta("nowInBlock", 1, 1),
+    "rowNumberInBlock": HogQLFunctionMeta("rowNumberInBlock", 0, 0),
     "rowNumberInAllBlocks": HogQLFunctionMeta("rowNumberInAllBlocks", 0, 0),
     "today": HogQLFunctionMeta("today"),
     "yesterday": HogQLFunctionMeta("yesterday"),
@@ -833,6 +834,10 @@ def compare_types(arg_types: list[ConstantType], sig_arg_types: tuple[ConstantTy
     "leadInFrame": HogQLFunctionMeta("leadInFrame", 1, 1),
     # table functions
     "generateSeries": HogQLFunctionMeta("generate_series", 3, 3),
+    ## UDFS
+    "aggregate_funnel": HogQLFunctionMeta("aggregate_funnel", 6, 6, aggregate=False),
+    "aggregate_funnel_array": HogQLFunctionMeta("aggregate_funnel_array", 6, 6, aggregate=False),
+    "aggregate_funnel_cohort": HogQLFunctionMeta("aggregate_funnel_cohort", 6, 6, aggregate=False),
 }
 # Permitted HogQL aggregations
 HOGQL_AGGREGATIONS: dict[str, HogQLFunctionMeta] = {
@@ -884,7 +889,7 @@ def compare_types(arg_types: list[ConstantType], sig_arg_types: tuple[ConstantTy
     # "topKWeighted": HogQLFunctionMeta("topKWeighted", 1, 1, aggregate=True),
     # "topKWeightedIf": HogQLFunctionMeta("topKWeightedIf", 2, 2, aggregate=True),
     "groupArray": HogQLFunctionMeta("groupArray", 1, 1, aggregate=True),
-    # "groupArrayIf": HogQLFunctionMeta("groupArrayIf", 2, 2, aggregate=True),
+    "groupArrayIf": HogQLFunctionMeta("groupArrayIf", 2, 2, aggregate=True),
     # "groupArrayLast": HogQLFunctionMeta("groupArrayLast", 1, 1, aggregate=True),
     # "groupArrayLastIf": HogQLFunctionMeta("groupArrayLastIf", 2, 2, aggregate=True),
     "groupUniqArray": HogQLFunctionMeta("groupUniqArray", 1, 1, aggregate=True),

posthog/hogql_queries/insights/funnels/__init__.py

@@ -1,5 +1,6 @@
 from .base import FunnelBase
 from .funnel import Funnel
+from .funnel_udf import FunnelUDF
 from .funnel_strict import FunnelStrict
 from .funnel_unordered import FunnelUnordered
 from .funnel_time_to_convert import FunnelTimeToConvert

posthog/hogql_queries/insights/funnels/base.py

@@ -33,6 +33,7 @@
     FunnelExclusionActionsNode,
     FunnelTimeToConvertResults,
     FunnelVizType,
+    FunnelExclusionEventsNode,
 )
 from posthog.types import EntityNode, ExclusionEntityNode
 
@@ -408,6 +409,93 @@ def _get_inner_event_query(
 
         return funnel_events_query
 
+    # This version of the inner event query modifies how exclusions are returned to
+    # make them behave more like steps. It returns a boolean "exclusion_{0..n}" for each event
+    def _get_inner_event_query_for_udf(
+        self,
+        entities: list[EntityNode] | None = None,
+        entity_name="events",
+        skip_entity_filter=False,
+        skip_step_filter=False,
+    ) -> ast.SelectQuery:
+        query, funnelsFilter, breakdown, breakdownType, breakdownAttributionType = (
+            self.context.query,
+            self.context.funnelsFilter,
+            self.context.breakdown,
+            self.context.breakdownType,
+            self.context.breakdownAttributionType,
+        )
+        entities_to_use = entities or query.series
+
+        extra_fields: list[str] = []
+
+        for prop in self.context.includeProperties:
+            extra_fields.append(prop)
+
+        funnel_events_query = FunnelEventQuery(
+            context=self.context,
+            extra_fields=[*self._extra_event_fields, *extra_fields],
+            extra_event_properties=self._extra_event_properties,
+        ).to_query(
+            skip_entity_filter=skip_entity_filter,
+        )
+        # funnel_events_query, params = FunnelEventQuery(
+        #     extra_fields=[*self._extra_event_fields, *extra_fields],
+        #     extra_event_properties=self._extra_event_properties,
+        # ).get_query(entities_to_use, entity_name, skip_entity_filter=skip_entity_filter)
+
+        all_step_cols: list[ast.Expr] = []
+        all_exclusions: list[list[FunnelExclusionEventsNode | FunnelExclusionActionsNode]] = []
+        for index, entity in enumerate(entities_to_use):
+            step_cols = self._get_step_col(entity, index, entity_name)
+            all_step_cols.extend(step_cols)
+            all_exclusions.append([])
+
+        for excluded_entity in funnelsFilter.exclusions or []:
+            for i in range(excluded_entity.funnelFromStep + 1, excluded_entity.funnelToStep + 1):
+                all_exclusions[i].append(excluded_entity)
+
+        for index, exclusions in enumerate(all_exclusions):
+            exclusion_col_expr = self._get_exclusions_col(exclusions, index, entity_name)
+            all_step_cols.append(exclusion_col_expr)
+
+        breakdown_select_prop = self._get_breakdown_select_prop()
+
+        if breakdown_select_prop:
+            all_step_cols.extend(breakdown_select_prop)
+
+        funnel_events_query.select = [*funnel_events_query.select, *all_step_cols]
+
+        if breakdown and breakdownType == BreakdownType.COHORT:
+            if funnel_events_query.select_from is None:
+                raise ValidationError("Apologies, there was an error adding cohort breakdowns to the query.")
+            funnel_events_query.select_from.next_join = self._get_cohort_breakdown_join()
+
+        if not skip_step_filter:
+            assert isinstance(funnel_events_query.where, ast.Expr)
+            steps_conditions = self._get_steps_conditions_for_udf(all_exclusions, length=len(entities_to_use))
+            funnel_events_query.where = ast.And(exprs=[funnel_events_query.where, steps_conditions])
+
+        if breakdown and breakdownAttributionType != BreakdownAttributionType.ALL_EVENTS:
+            # ALL_EVENTS attribution is the old default, which doesn't need the subquery
+            return self._add_breakdown_attribution_subquery(funnel_events_query)
+
+        return funnel_events_query
+
+    def _get_exclusions_col(
+        self,
+        exclusions: list[ExclusionEntityNode],
+        index: int,
+        entity_name: str,
+    ) -> ast.Expr:
+        if not exclusions:
+            return parse_expr(f"0 as exclusion_{index}")
+
+        conditions = [self._build_step_query(exclusion, index, entity_name, "") for exclusion in exclusions]
+        return parse_expr(
+            f"if({{condition}}, 1, 0) as exclusion_{index}", placeholders={"condition": ast.Or(exprs=conditions)}
+        )
+
     def _get_cohort_breakdown_join(self) -> ast.JoinExpr:
         breakdown = self.context.breakdown
 
@@ -545,12 +633,23 @@ def _get_steps_conditions(self, length: int) -> ast.Expr:
 
         return ast.Or(exprs=step_conditions)
 
+    def _get_steps_conditions_for_udf(self, exclusions, length: int) -> ast.Expr:
+        step_conditions: list[ast.Expr] = []
+
+        for index in range(length):
+            step_conditions.append(parse_expr(f"step_{index} = 1"))
+            if exclusions[index]:
+                step_conditions.append(parse_expr(f"exclusion_{index} = 1"))
+
+        return ast.Or(exprs=step_conditions)
+
     def _get_step_col(
         self,
         entity: EntityNode | ExclusionEntityNode,
         index: int,
         entity_name: str,
         step_prefix: str = "",
+        for_udf: bool = False,
     ) -> list[ast.Expr]:
         # step prefix is used to distinguish actual steps, and exclusion steps
         # without the prefix, we get the same parameter binding for both, which borks things up
@@ -559,9 +658,10 @@ def _get_step_col(
         step_cols.append(
             parse_expr(f"if({{condition}}, 1, 0) as {step_prefix}step_{index}", placeholders={"condition": condition})
         )
-        step_cols.append(
-            parse_expr(f"if({step_prefix}step_{index} = 1, timestamp, null) as {step_prefix}latest_{index}")
-        )
+        if not for_udf:
+            step_cols.append(
+                parse_expr(f"if({step_prefix}step_{index} = 1, timestamp, null) as {step_prefix}latest_{index}")
+            )
 
         for field in self.extra_event_fields_and_properties:
             step_cols.append(

posthog/hogql_queries/insights/funnels/funnel_time_to_convert.py

@@ -16,7 +16,8 @@ def __init__(
     ):
         super().__init__(context)
 
-        self.funnel_order = get_funnel_order_class(self.context.funnelsFilter)(context=self.context)
+        # Haven't implemented calls for time_to_convert in UDF yet
+        self.funnel_order = get_funnel_order_class(self.context.funnelsFilter, use_udf=False)(context=self.context)
 
     def _format_results(self, results: list) -> FunnelTimeToConvertResults:
         return FunnelTimeToConvertResults(