feat(experiments): Support for data warehouse experiments

ee/clickhouse/views/experiments.py

@@ -339,6 +339,7 @@ def update(self, instance: Experiment, validated_data: dict, *args: Any, **kwarg
         # if (
         #     not instance.filters.get("events")
         #     and not instance.filters.get("actions")
+        #     and not instance.filters.get("data_warehouse")
         #     and validated_data.get("start_date")
         #     and not validated_data.get("filters")
         # ):

frontend/src/scenes/experiments/ExperimentView/Goal.tsx

@@ -73,7 +73,13 @@ export function MetricDisplayOld({ filters }: { filters?: FilterType }): JSX.Ele
 
     return (
         <>
-            {([...(filters?.events || []), ...(filters?.actions || [])] as ActionFilter[])
+            {(
+                [
+                    ...(filters?.events || []),
+                    ...(filters?.actions || []),
+                    ...(filters?.data_warehouse || []),
+                ] as ActionFilter[]
+            )
                 .sort((a, b) => (a.order || 0) - (b.order || 0))
                 .map((event: ActionFilter, idx: number) => (
                     <div key={idx} className="mb-2">

frontend/src/scenes/experiments/experimentLogic.tsx

@@ -652,7 +652,8 @@ export const experimentLogic = kea<experimentLogicType>([
         setExperiment: async ({ experiment }) => {
             const experimentEntitiesChanged =
                 (experiment.filters?.events && experiment.filters.events.length > 0) ||
-                (experiment.filters?.actions && experiment.filters.actions.length > 0)
+                (experiment.filters?.actions && experiment.filters.actions.length > 0) ||
+                (experiment.filters?.data_warehouse && experiment.filters.data_warehouse.length > 0)
 
             if (!experiment.filters || Object.keys(experiment.filters).length === 0) {
                 return
@@ -667,7 +668,9 @@ export const experimentLogic = kea<experimentLogicType>([
 
             if (name === 'filters') {
                 const experimentEntitiesChanged =
-                    (value?.events && value.events.length > 0) || (value?.actions && value.actions.length > 0)
+                    (value?.events && value.events.length > 0) ||
+                    (value?.actions && value.actions.length > 0) ||
+                    (value?.data_warehouse && value.data_warehouse.length > 0)
 
                 if (!value || Object.keys(value).length === 0) {
                     return
@@ -685,7 +688,8 @@ export const experimentLogic = kea<experimentLogicType>([
 
             const experimentEntitiesChanged =
                 (experiment.filters?.events && experiment.filters.events.length > 0) ||
-                (experiment.filters?.actions && experiment.filters.actions.length > 0)
+                (experiment.filters?.actions && experiment.filters.actions.length > 0) ||
+                (experiment.filters?.data_warehouse && experiment.filters.data_warehouse.length > 0)
 
             if (!experiment.filters || Object.keys(experiment.filters).length === 0) {
                 return
@@ -699,7 +703,8 @@ export const experimentLogic = kea<experimentLogicType>([
             const experiment = values.experiment
             const experimentEntitiesChanged =
                 (experiment.filters?.events && experiment.filters.events.length > 0) ||
-                (experiment.filters?.actions && experiment.filters.actions.length > 0)
+                (experiment.filters?.actions && experiment.filters.actions.length > 0) ||
+                (experiment.filters?.data_warehouse && experiment.filters.data_warehouse.length > 0)
 
             if (!experiment.filters || Object.keys(experiment.filters).length === 0) {
                 return
@@ -1045,7 +1050,11 @@ export const experimentLogic = kea<experimentLogicType>([
                     if (!filters) {
                         return undefined
                     }
-                    entities = [...(filters?.events || []), ...(filters?.actions || [])] as ActionFilterType[]
+                    entities = [
+                        ...(filters?.events || []),
+                        ...(filters?.actions || []),
+                        ...(filters?.data_warehouse || []),
+                    ] as ActionFilterType[]
                 }
 
                 // Find out if we're using count per actor math aggregates averages per user

posthog/hogql_queries/experiments/experiment_trends_query_runner.py

@@ -3,6 +3,9 @@
 from django.conf import settings
 from posthog.constants import ExperimentNoResultsErrorKeys
 from posthog.hogql import ast
+from posthog.hogql.database.database import create_hogql_database
+from posthog.hogql.database.models import LazyJoin
+from posthog.hogql.database.schema.events import EventsTable
 from posthog.hogql_queries.experiments import CONTROL_VARIANT_KEY
 from posthog.hogql_queries.experiments.trends_statistics import (
     are_results_significant,
@@ -19,6 +22,8 @@
     BreakdownFilter,
     CachedExperimentTrendsQueryResponse,
     ChartDisplayType,
+    DataWarehouseNode,
+    DataWarehousePropertyFilter,
     EventPropertyFilter,
     EventsNode,
     ExperimentSignificanceCode,
@@ -27,6 +32,7 @@
     ExperimentVariantTrendsBaseStats,
     InsightDateRange,
     PropertyMathType,
+    PropertyOperator,
     TrendsFilter,
     TrendsQuery,
     TrendsQueryResponse,
@@ -89,12 +95,18 @@
             explicitDate=True,
         )
 
-    def _get_breakdown_filter(self) -> BreakdownFilter:
+    def _get_event_breakdown_filter(self) -> BreakdownFilter:
         return BreakdownFilter(
             breakdown=self.breakdown_key,
             breakdown_type="event",
         )
 
+    def _get_data_warehouse_breakdown_filter(self, column_name: str) -> BreakdownFilter:
+        return BreakdownFilter(
+            breakdown=f"{column_name}.properties.`{self.breakdown_key}`",
+            breakdown_type="data_warehouse",
+        )
+
     def _prepare_count_query(self) -> TrendsQuery:
         """
         This method takes the raw trend query and adapts it
@@ -109,6 +121,10 @@
 
         uses_math_aggregation = self._uses_math_aggregation_by_user_or_property_value(prepared_count_query)
 
+        # TODO:
+        # - Apply the column name to the breakdown filter and transform the breakdown type to datawarehouse_property
+        # - Apply the column name to the properties filter and transform the property type to datawarehouse property
+
         # :TRICKY: for `avg` aggregation, use `sum` data as an approximation
         if prepared_count_query.series[0].math == PropertyMathType.AVG:
             prepared_count_query.series[0].math = PropertyMathType.SUM
@@ -118,15 +134,35 @@
 
         prepared_count_query.trendsFilter = TrendsFilter(display=ChartDisplayType.ACTIONS_LINE_GRAPH_CUMULATIVE)
         prepared_count_query.dateRange = self._get_insight_date_range()
-        prepared_count_query.breakdownFilter = self._get_breakdown_filter()
-        prepared_count_query.properties = [
-            EventPropertyFilter(
-                key=self.breakdown_key,
-                value=self.variants,
-                operator="exact",
-                type="event",
-            )
-        ]
+        if self._is_data_warehouse_query(prepared_count_query):
+            column_name = self._get_data_warehouse_events_column_name(prepared_count_query)
+            if not column_name:
+                raise ValueError("Data warehouse table is expected to have a lazy join to the events table")
+            prepared_count_query.breakdownFilter = self._get_data_warehouse_breakdown_filter(column_name)
+            prepared_count_query.properties = [
+                DataWarehousePropertyFilter(
+                    key=f"{column_name}.event",
+                    value="$feature_flag_called",
+                    operator=PropertyOperator.EXACT,
+                    type="data_warehouse",
+                ),
+                DataWarehousePropertyFilter(
+                    key=f"{column_name}.properties.`{self.breakdown_key}`",
+                    value=self.variants,
+                    operator=PropertyOperator.EXACT,
+                    type="data_warehouse",
+                ),
+            ]
+        else:
+            prepared_count_query.breakdownFilter = self._get_event_breakdown_filter()
+            prepared_count_query.properties = [
+                EventPropertyFilter(
+                    key=self.breakdown_key,
+                    value=self.variants,
+                    operator=PropertyOperator.EXACT,
+                    type="event",
+                )
+            ]
 
         return prepared_count_query
 
@@ -152,7 +188,7 @@
 
             if hasattr(count_event, "event"):
                 prepared_exposure_query.dateRange = self._get_insight_date_range()
-                prepared_exposure_query.breakdownFilter = self._get_breakdown_filter()
+                prepared_exposure_query.breakdownFilter = self._get_event_breakdown_filter()
                 prepared_exposure_query.trendsFilter = TrendsFilter(
                     display=ChartDisplayType.ACTIONS_LINE_GRAPH_CUMULATIVE
                 )
@@ -166,7 +202,7 @@
                     EventPropertyFilter(
                         key=self.breakdown_key,
                         value=self.variants,
-                        operator="exact",
+                        operator=PropertyOperator.EXACT,
                         type="event",
                     )
                 ]
@@ -177,16 +213,30 @@
         elif self.query.exposure_query:
             prepared_exposure_query = TrendsQuery(**self.query.exposure_query.model_dump())
             prepared_exposure_query.dateRange = self._get_insight_date_range()
-            prepared_exposure_query.breakdownFilter = self._get_breakdown_filter()
             prepared_exposure_query.trendsFilter = TrendsFilter(display=ChartDisplayType.ACTIONS_LINE_GRAPH_CUMULATIVE)
-            prepared_exposure_query.properties = [
-                EventPropertyFilter(
-                    key=self.breakdown_key,
-                    value=self.variants,
-                    operator="exact",
-                    type="event",
-                )
-            ]
+            if self._is_data_warehouse_query(prepared_exposure_query):
+                column_name = self._get_data_warehouse_events_column_name(prepared_exposure_query)
+                if not column_name:
+                    raise ValueError("Data warehouse table is expected to have a lazy join to the events table")
+                prepared_exposure_query.breakdownFilter = self._get_data_warehouse_breakdown_filter(column_name)
+                prepared_exposure_query.properties = [
+                    DataWarehousePropertyFilter(
+                        key=f"properties.`{self.breakdown_key}`",
+                        value=self.variants,
+                        operator=PropertyOperator.EXACT,
+                        type="data_warehouse",
+                    )
+                ]
+            else:
+                prepared_exposure_query.breakdownFilter = self._get_event_breakdown_filter()
+                prepared_exposure_query.properties = [
+                    EventPropertyFilter(
+                        key=self.breakdown_key,
+                        value=self.variants,
+                        operator=PropertyOperator.EXACT,
+                        type="event",
+                    )
+                ]
         # 3. Otherwise, we construct a default exposure query: unique users for the $feature_flag_called event
         else:
             prepared_exposure_query = TrendsQuery(
@@ -206,13 +256,13 @@
                     EventPropertyFilter(
                         key="$feature_flag_response",
                         value=self.variants,
-                        operator="exact",
+                        operator=PropertyOperator.EXACT,
                         type="event",
                     ),
                     EventPropertyFilter(
                         key="$feature_flag",
                         value=[self.feature_flag.key],
-                        operator="exact",
+                        operator=PropertyOperator.EXACT,
                         type="event",
                     ),
                 ],
@@ -362,5 +412,38 @@
         if has_errors:
             raise ValidationError(detail=json.dumps(errors))
 
+    def _is_data_warehouse_query(self, query: TrendsQuery) -> bool:
+        return any(isinstance(series, DataWarehouseNode) for series in query.series)
+
+    def _get_data_warehouse_events_column_name(self, query: TrendsQuery) -> Optional[str]:
+        if hasattr(self, "data_warehouse_events_column_name"):
+            return self.data_warehouse_events_column_name
+
+        self.data_warehouse_events_column_name = None
+
+        try:
+            if not query.series:
+                raise ValueError("Query series is empty")
+
+            database = create_hogql_database(self.team.pk)
+            if not database:
+                raise ValueError("Failed to create HogQL database")
+
+            table_name = query.series[0].table_name
+            if table_name not in database.model_extra:
+                raise ValueError(f"Table '{table_name}' not found in database model")
+
+            table_fields = database.model_extra[table_name].fields
+            for field_name, field in table_fields.items():
+                if isinstance(field, LazyJoin) and isinstance(field.join_table, EventsTable):
+                    self.data_warehouse_events_column_name = field_name
+                    break
+
+            return self.data_warehouse_events_column_name
+
+        except Exception as e:
+            # You might want to log the error here
+            raise ValueError(f"Failed to get data warehouse events column name: {str(e)}") from e
+
     def to_query(self) -> ast.SelectQuery:
         raise ValueError(f"Cannot convert source query of type {self.query.count_query.kind} to query")

posthog/hogql_queries/insights/trends/trends_query_runner.py

@@ -888,7 +888,16 @@ def _is_breakdown_filter_field_boolean(self):
             if not table_or_view:
                 raise ValueError(f"Table {series.table_name} not found")
 
-            field_type = dict(table_or_view.columns)[self.query.breakdownFilter.breakdown]["clickhouse"]
+            breakdown_key = (
+                self.query.breakdownFilter.breakdown[0]
+                if isinstance(self.query.breakdownFilter.breakdown, list)
+                else self.query.breakdownFilter.breakdown
+            )
+
+            if breakdown_key not in dict(table_or_view.columns):
+                return False
+
+            field_type = dict(table_or_view.columns)[breakdown_key]["clickhouse"]
 
             if field_type.startswith("Nullable("):
                 field_type = field_type.replace("Nullable(", "")[:-1]

posthog/hogql_queries/insights/trends/utils.py

@@ -26,7 +26,7 @@ def get_properties_chain(
         raise Exception("group_type_index missing from params")
 
     if breakdown_type == "data_warehouse":
-        return [breakdown_field]
+        return [*breakdown_field.split(".")]
 
     if breakdown_type == "data_warehouse_person_property":
         return ["person", *breakdown_field.split(".")]