feat(data-warehouse): integrating data warehouse with trends insight

posthog/constants.py

@@ -40,6 +40,7 @@ class AvailableFeature(str, Enum):
 
 TREND_FILTER_TYPE_ACTIONS = "actions"
 TREND_FILTER_TYPE_EVENTS = "events"
+TREND_FILTER_TYPE_DATA_WAREHOUSE = "data_warehouse"
 
 SESSION_RECORDINGS_FILTER_IDS = "session_ids"
 
@@ -117,6 +118,7 @@ class AvailableFeature(str, Enum):
 ENTITIES = "entities"
 ACTIONS = "actions"
 EVENTS = "events"
+DATA_WAREHOUSE_ENTITIES = "data_warehouse_entities"
 EXCLUSIONS = "exclusions"
 PROPERTIES = "properties"
 PROPERTY_GROUPS = "property_groups"

posthog/hogql/functions/mapping.py

@@ -169,7 +169,7 @@ class HogQLFunctionMeta:
         "parseDateTime64BestEffortOrNull",
         1,
         1,
-        overloads=[((ast.DateTimeType, ast.DateType), "toDateTime")],
+        overloads=[((ast.DateTimeType, ast.DateType, ast.IntegerType), "toDateTime")],
         tz_aware=True,
     ),
     "toUUID": HogQLFunctionMeta("toUUIDOrNull", 1, 1),

posthog/hogql/property.py

@@ -121,7 +121,11 @@
     if property.type == "hogql":
         return parse_expr(property.key)
     elif (
-        property.type == "event" or property.type == "feature" or property.type == "person" or property.type == "group"
+        property.type == "event"
+        or property.type == "feature"
+        or property.type == "person"
+        or property.type == "group"
+        or property.type == "data_warehouse"
     ):
         if scope == "person" and property.type != "person":
             raise NotImplementedException(
@@ -134,6 +138,8 @@
             chain = ["person", "properties"]
         elif property.type == "group":
             chain = [f"group_{property.group_type_index}", "properties"]
+        elif property.type == "data_warehouse":
+            chain = []
         else:
             chain = ["properties"]
         field = ast.Field(chain=chain + [property.key])

posthog/hogql_queries/insights/trends/aggregation_operations.py

@@ -3,7 +3,8 @@
 from posthog.hogql.parser import parse_expr, parse_select
 from posthog.hogql_queries.utils.query_date_range import QueryDateRange
 from posthog.models.team.team import Team
-from posthog.schema import ActionsNode, EventsNode
+from posthog.schema import SeriesType, DataWarehouseNode
+from posthog.models.filters.mixins.utils import cached_property
 
 
 class QueryAlternator:
@@ -49,14 +50,14 @@ def replace_select_from(self, join_expr: ast.JoinExpr) -> None:
 
 class AggregationOperations:
     team: Team
-    series: EventsNode | ActionsNode
+    series: SeriesType
     query_date_range: QueryDateRange
     should_aggregate_values: bool
 
     def __init__(
         self,
         team: Team,
-        series: EventsNode | ActionsNode,
+        series: SeriesType,
         query_date_range: QueryDateRange,
         should_aggregate_values: bool,
     ) -> None:
@@ -65,11 +66,18 @@ def __init__(
         self.query_date_range = query_date_range
         self.should_aggregate_values = should_aggregate_values
 
+    @cached_property
+    def _id_field(self) -> ast.Expr:
+        if isinstance(self.series, DataWarehouseNode):
+            return ast.Field(chain=["e", self.series.id_field])
+
+        return ast.Field(chain=["e", "uuid"])
+
     def select_aggregation(self) -> ast.Expr:
         if self.series.math == "hogql" and self.series.math_hogql is not None:
             return parse_expr(self.series.math_hogql)
         elif self.series.math == "total":
-            return parse_expr("count(e.uuid)")
+            return parse_expr("count({id_field})", placeholders={"id_field": self._id_field})
         elif self.series.math == "dau":
             actor = "e.distinct_id" if self.team.aggregate_users_by_distinct_id else "e.person.id"
             return parse_expr(f"count(DISTINCT {actor})")
@@ -99,7 +107,9 @@ def select_aggregation(self) -> ast.Expr:
             elif self.series.math == "p99":
                 return self._math_quantile(0.99, None)
 
-        return parse_expr("count(e.uuid)")  # All "count per actor" get replaced during query orchestration
+        return parse_expr(
+            "count({id_field})", placeholders={"id_field": self._id_field}
+        )  # All "count per actor" get replaced during query orchestration
 
     def requires_query_orchestration(self) -> bool:
         math_to_return_true = [
@@ -330,13 +340,14 @@ def _events_query(
             query = parse_select(
                 """
                     SELECT
-                        count(e.uuid) AS total
+                        count({id_field}) AS total
                     FROM events AS e
                     SAMPLE {sample}
                     WHERE {events_where_clause}
                     GROUP BY {person_field}
                 """,
                 placeholders={
+                    "id_field": self._id_field,
                     "events_where_clause": where_clause_combined,
                     "sample": sample_value,
                     "person_field": ast.Field(

posthog/hogql_queries/insights/trends/breakdown.py

@@ -12,18 +12,17 @@
 from posthog.hogql_queries.insights.trends.display import TrendsDisplay
 from posthog.hogql_queries.insights.trends.utils import (
     get_properties_chain,
-    series_event_name,
 )
 from posthog.hogql_queries.utils.query_date_range import QueryDateRange
 from posthog.models.filters.mixins.utils import cached_property
 from posthog.models.team.team import Team
-from posthog.schema import ActionsNode, EventsNode, HogQLQueryModifiers, InCohortVia, TrendsQuery
+from posthog.schema import SeriesType, HogQLQueryModifiers, InCohortVia, TrendsQuery
 
 
 class Breakdown:
     query: TrendsQuery
     team: Team
-    series: EventsNode | ActionsNode
+    series: SeriesType
     query_date_range: QueryDateRange
     timings: HogQLTimings
     modifiers: HogQLQueryModifiers
@@ -34,7 +33,7 @@
         self,
         team: Team,
         query: TrendsQuery,
-        series: EventsNode | ActionsNode,
+        series: SeriesType,
         query_date_range: QueryDateRange,
         timings: HogQLTimings,
         modifiers: HogQLQueryModifiers,
@@ -222,7 +221,7 @@
         with self.timings.measure("breakdown_values_query"):
             breakdown = BreakdownValues(
                 team=self.team,
-                event_name=series_event_name(self.series) or "",
+                series=self.series,
                 events_filter=self.events_filter,
                 chart_display_type=self._trends_display().display_type,
                 breakdown_filter=self.query.breakdownFilter,
@@ -281,7 +280,7 @@
     @cached_property
     def _properties_chain(self):
         return get_properties_chain(
             breakdown_type=self.query.breakdownFilter.breakdown_type,
             breakdown_field=self.query.breakdownFilter.breakdown,
             group_type_index=self.query.breakdownFilter.breakdown_group_type_index,
         )

posthog/hogql_queries/insights/trends/breakdown_values.py

@@ -5,7 +5,8 @@
 from posthog.hogql.query import execute_hogql_query
 from posthog.hogql_queries.insights.trends.utils import get_properties_chain
 from posthog.models.team.team import Team
-from posthog.schema import BreakdownFilter, BreakdownType, ChartDisplayType
+from posthog.schema import BreakdownFilter, BreakdownType, ChartDisplayType, SeriesType, DataWarehouseNode
+from functools import cached_property
 
 BREAKDOWN_OTHER_STRING_LABEL = "$$_posthog_breakdown_other_$$"
 BREAKDOWN_OTHER_NUMERIC_LABEL = 9007199254740991  # pow(2, 53) - 1, for JS compatibility
@@ -15,7 +16,7 @@
 
 class BreakdownValues:
     team: Team
-    event_name: str
+    series: SeriesType
     breakdown_field: Union[str, float, List[Union[str, float]]]
     breakdown_type: BreakdownType
     events_filter: ast.Expr
@@ -28,13 +29,13 @@ class BreakdownValues:
     def __init__(
         self,
         team: Team,
-        event_name: str,
+        series: SeriesType,
         events_filter: ast.Expr,
         chart_display_type: ChartDisplayType,
         breakdown_filter: BreakdownFilter,
     ):
         self.team = team
-        self.event_name = event_name
+        self.series = series
         self.breakdown_field = breakdown_filter.breakdown  # type: ignore
         self.breakdown_type = breakdown_filter.breakdown_type  # type: ignore
         self.events_filter = events_filter
@@ -88,9 +89,8 @@ def get_breakdown_values(self) -> List[str | int]:
             """
                 SELECT
                     {select_field},
-                    count(e.uuid) as count
-                FROM
-                    events e
+                    count({id_field}) as count
+                FROM {table} e
                 WHERE
                     {events_where}
                 GROUP BY
@@ -104,6 +104,8 @@ def get_breakdown_values(self) -> List[str | int]:
                 "events_where": self.events_filter,
                 "select_field": select_field,
                 "breakdown_limit": ast.Constant(value=breakdown_limit),
+                "table": self._table,
+                "id_field": self._id_field,
             },
         )
 
@@ -166,3 +168,17 @@ def _to_bucketing_expression(self) -> ast.Expr:
             qunatile_expression = f"quantiles({','.join([f'{quantile:.2f}' for quantile in quantiles])})(value)"
 
         return parse_expr(f"arrayCompact(arrayMap(x -> floor(x, 2), {qunatile_expression}))")
+
+    @cached_property
+    def _id_field(self) -> ast.Field:
+        if isinstance(self.series, DataWarehouseNode):
+            return ast.Field(chain=["e", self.series.id_field])
+
+        return ast.Field(chain=["e", "uuid"])
+
+    @cached_property
+    def _table(self) -> ast.Field:
+        if isinstance(self.series, DataWarehouseNode):
+            return ast.Field(chain=[self.series.table_name])
+
+        return ast.Field(chain=["events"])