From 9f52e061eb5bd3d5a4aa86e4f15a1f00393fc362 Mon Sep 17 00:00:00 2001
From: Eric Duong <eeoneric@gmail.com>
Date: Fri, 15 Mar 2024 12:13:13 -0400
Subject: [PATCH] feat(data-warehouse): virtual data warehouse table (#20888)

* create virtual table definition with the shape of events

* add conditionals

* update schema

* add distinct_id field

* maths working

* add prop math selection

* add field

* Update UI snapshots for `chromium` (2)

* Update UI snapshots for `chromium` (2)

* typing

* clean up snapshot

* cleanup

* remove unnecessary

* more cleanup

* add breakdown

* format

* typing

---------

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>
---
 .../DefinitionPopoverContents.tsx             |  17 +-
 .../TaxonomicPopover/TaxonomicPopover.tsx     |   5 +-
 .../InsightQuery/utils/filtersToQueryNode.ts  |   2 +
 .../InsightQuery/utils/queryNodeToFilter.ts   |   7 +-
 frontend/src/queries/schema.json              |  30 +-
 frontend/src/queries/schema.ts                |   9 +
 frontend/src/scenes/data-warehouse/types.ts   |   1 +
 .../ActionFilterRow/ActionFilterRow.tsx       |  10 +-
 .../filters/ActionFilter/entityFilterLogic.ts |  19 +-
 frontend/src/types.ts                         |   1 +
 posthog/hogql/database/database.py            |  28 ++
 .../insights/data_warehouse_mixin.py          |  14 +
 .../insights/trends/aggregation_operations.py |  14 +-
 .../insights/trends/breakdown.py              |   1 +
 .../insights/trends/breakdown_values.py       |  15 +-
 .../data_warehouse_trends_query_builder.py    | 409 ------------------
 .../test_data_warehouse_query_builder.ambr    |  26 --
 .../test_trends_data_warehouse_query.ambr     | 185 ++++++++
 ...py => test_trends_data_warehouse_query.py} |  68 ++-
 .../insights/trends/trends_query_builder.py   |  37 +-
 .../trends/trends_query_builder_abstract.py   |  52 ---
 .../insights/trends/trends_query_runner.py    |  45 +-
 posthog/schema.py                             |  12 +
 23 files changed, 468 insertions(+), 539 deletions(-)
 create mode 100644 posthog/hogql_queries/insights/data_warehouse_mixin.py
 delete mode 100644 posthog/hogql_queries/insights/trends/data_warehouse_trends_query_builder.py
 create mode 100644 posthog/hogql_queries/insights/trends/test/__snapshots__/test_trends_data_warehouse_query.ambr
 rename posthog/hogql_queries/insights/trends/test/{test_data_warehouse_query_builder.py => test_trends_data_warehouse_query.py} (81%)
 delete mode 100644 posthog/hogql_queries/insights/trends/trends_query_builder_abstract.py

diff --git a/frontend/src/lib/components/DefinitionPopover/DefinitionPopoverContents.tsx b/frontend/src/lib/components/DefinitionPopover/DefinitionPopoverContents.tsx
index e174b688a36f6..1c9e4928ecb22 100644
--- a/frontend/src/lib/components/DefinitionPopover/DefinitionPopoverContents.tsx
+++ b/frontend/src/lib/components/DefinitionPopover/DefinitionPopoverContents.tsx
@@ -293,7 +293,18 @@ function DefinitionView({ group }: { group: TaxonomicFilterGroup }): JSX.Element
                             onChange={(value) => setLocalDefinition({ id_field: value })}
                         />
 
-                        <label className="definition-popover-edit-form-label" htmlFor="ID Field">
+                        <label className="definition-popover-edit-form-label" htmlFor="Distinct Id Field">
+                            <span className="label-text">Distinct ID field</span>
+                        </label>
+                        <LemonSelect
+                            value={
+                                'distinct_id_field' in localDefinition ? localDefinition.distinct_id_field : undefined
+                            }
+                            options={columnOptions}
+                            onChange={(value) => setLocalDefinition({ distinct_id_field: value })}
+                        />
+
+                        <label className="definition-popover-edit-form-label" htmlFor="Timestamp Field">
                             <span className="label-text">Timestamp field</span>
                         </label>
                         <LemonSelect
@@ -313,7 +324,9 @@ function DefinitionView({ group }: { group: TaxonomicFilterGroup }): JSX.Element
                                 'id_field' in localDefinition &&
                                 localDefinition.id_field &&
                                 'timestamp_field' in localDefinition &&
-                                localDefinition.timestamp_field
+                                localDefinition.timestamp_field &&
+                                'distinct_id_field' in localDefinition &&
+                                localDefinition.distinct_id_field
                                     ? null
                                     : 'Field mappings must be specified'
                             }
diff --git a/frontend/src/lib/components/TaxonomicPopover/TaxonomicPopover.tsx b/frontend/src/lib/components/TaxonomicPopover/TaxonomicPopover.tsx
index 9f4a5d537edf3..4fe515646323c 100644
--- a/frontend/src/lib/components/TaxonomicPopover/TaxonomicPopover.tsx
+++ b/frontend/src/lib/components/TaxonomicPopover/TaxonomicPopover.tsx
@@ -5,7 +5,7 @@ import { LemonButton, LemonButtonProps } from 'lib/lemon-ui/LemonButton'
 import { LemonDropdown } from 'lib/lemon-ui/LemonDropdown'
 import { useEffect, useState } from 'react'
 
-import { AnyDataNode } from '~/queries/schema'
+import { AnyDataNode, DatabaseSchemaQueryResponseField } from '~/queries/schema'
 
 export interface TaxonomicPopoverProps<ValueType extends TaxonomicFilterValue = TaxonomicFilterValue>
     extends Omit<LemonButtonProps, 'children' | 'onClick' | 'sideIcon' | 'sideAction'> {
@@ -19,6 +19,7 @@ export interface TaxonomicPopoverProps<ValueType extends TaxonomicFilterValue =
     placeholder?: React.ReactNode
     placeholderClass?: string
     dropdownMatchSelectWidth?: boolean
+    schemaColumns?: DatabaseSchemaQueryResponseField[]
     allowClear?: boolean
     style?: React.CSSProperties
     excludedProperties?: { [key in TaxonomicFilterGroupType]?: TaxonomicFilterValue[] }
@@ -49,6 +50,7 @@ export function TaxonomicPopover<ValueType extends TaxonomicFilterValue = Taxono
     allowClear = false,
     excludedProperties,
     metadataSource,
+    schemaColumns,
     ...buttonPropsRest
 }: TaxonomicPopoverProps<ValueType>): JSX.Element {
     const [localValue, setLocalValue] = useState<ValueType>(value || ('' as ValueType))
@@ -85,6 +87,7 @@ export function TaxonomicPopover<ValueType extends TaxonomicFilterValue = Taxono
                     }}
                     taxonomicGroupTypes={groupTypes ?? [groupType]}
                     eventNames={eventNames}
+                    schemaColumns={schemaColumns}
                     metadataSource={metadataSource}
                     excludedProperties={excludedProperties}
                 />
diff --git a/frontend/src/queries/nodes/InsightQuery/utils/filtersToQueryNode.ts b/frontend/src/queries/nodes/InsightQuery/utils/filtersToQueryNode.ts
index 820c44f3fd12c..40cd113a7d29d 100644
--- a/frontend/src/queries/nodes/InsightQuery/utils/filtersToQueryNode.ts
+++ b/frontend/src/queries/nodes/InsightQuery/utils/filtersToQueryNode.ts
@@ -95,6 +95,7 @@ export const legacyEntityToNode = (
         custom_name: entity.custom_name || undefined,
         id_field: 'id_field' in entity ? entity.id_field : undefined,
         timestamp_field: 'timestamp_field' in entity ? entity.timestamp_field : undefined,
+        distinct_id_field: 'distinct_id_field' in entity ? entity.distinct_id_field : undefined,
         table_name: 'table_name' in entity ? entity.table_name : undefined,
     }
 
@@ -103,6 +104,7 @@ export const legacyEntityToNode = (
             ...shared,
             id_field: entity.id_field || undefined,
             timestamp_field: entity.timestamp_field || undefined,
+            distinct_id_field: entity.distinct_id_field || undefined,
             table_name: entity.table_name || undefined,
         }
     }
diff --git a/frontend/src/queries/nodes/InsightQuery/utils/queryNodeToFilter.ts b/frontend/src/queries/nodes/InsightQuery/utils/queryNodeToFilter.ts
index 7aff04beeb713..8c03814ead320 100644
--- a/frontend/src/queries/nodes/InsightQuery/utils/queryNodeToFilter.ts
+++ b/frontend/src/queries/nodes/InsightQuery/utils/queryNodeToFilter.ts
@@ -57,7 +57,12 @@ export const seriesNodeToFilter = (
         math_group_type_index: node.math_group_type_index,
         properties: node.properties as any, // TODO,
         ...(isDataWarehouseNode(node)
-            ? { table_name: node.table_name, id_field: node.id_field, timestamp_field: node.timestamp_field }
+            ? {
+                  table_name: node.table_name,
+                  id_field: node.id_field,
+                  timestamp_field: node.timestamp_field,
+                  distinct_id_field: node.distinct_id_field,
+              }
             : {}),
     })
     return entity
diff --git a/frontend/src/queries/schema.json b/frontend/src/queries/schema.json
index 843fa033737a2..898d4e0974004 100644
--- a/frontend/src/queries/schema.json
+++ b/frontend/src/queries/schema.json
@@ -764,12 +764,34 @@
             "required": ["kind", "source"],
             "type": "object"
         },
+        "DataWarehouseEventsModifier": {
+            "additionalProperties": false,
+            "properties": {
+                "distinct_id_field": {
+                    "type": "string"
+                },
+                "id_field": {
+                    "type": "string"
+                },
+                "table_name": {
+                    "type": "string"
+                },
+                "timestamp_field": {
+                    "type": "string"
+                }
+            },
+            "required": ["table_name", "timestamp_field", "distinct_id_field", "id_field"],
+            "type": "object"
+        },
         "DataWarehouseNode": {
             "additionalProperties": false,
             "properties": {
                 "custom_name": {
                     "type": "string"
                 },
+                "distinct_id_field": {
+                    "type": "string"
+                },
                 "fixedProperties": {
                     "description": "Fixed properties in the query, can't be edited in the interface (e.g. scoping down by person)",
                     "items": {
@@ -837,7 +859,7 @@
                     "type": "string"
                 }
             },
-            "required": ["id", "id_field", "kind", "table_name", "timestamp_field"],
+            "required": ["distinct_id_field", "id", "id_field", "kind", "table_name", "timestamp_field"],
             "type": "object"
         },
         "DataWarehousePropertyFilter": {
@@ -2526,6 +2548,12 @@
             "additionalProperties": false,
             "description": "HogQL Query Options are automatically set per team. However, they can be overriden in the query.",
             "properties": {
+                "dataWarehouseEventsModifiers": {
+                    "items": {
+                        "$ref": "#/definitions/DataWarehouseEventsModifier"
+                    },
+                    "type": "array"
+                },
                 "inCohortVia": {
                     "enum": ["auto", "leftjoin", "subquery", "leftjoin_conjoined"],
                     "type": "string"
diff --git a/frontend/src/queries/schema.ts b/frontend/src/queries/schema.ts
index 64de744e22d71..7eaa1734ea463 100644
--- a/frontend/src/queries/schema.ts
+++ b/frontend/src/queries/schema.ts
@@ -180,6 +180,14 @@ export interface HogQLQueryModifiers {
     personsArgMaxVersion?: 'auto' | 'v1' | 'v2'
     inCohortVia?: 'auto' | 'leftjoin' | 'subquery' | 'leftjoin_conjoined'
     materializationMode?: 'auto' | 'legacy_null_as_string' | 'legacy_null_as_null' | 'disabled'
+    dataWarehouseEventsModifiers?: DataWarehouseEventsModifier[]
+}
+
+export interface DataWarehouseEventsModifier {
+    table_name: string
+    timestamp_field: string
+    distinct_id_field: string
+    id_field: string
 }
 
 export interface HogQLQueryResponse {
@@ -376,6 +384,7 @@ export interface DataWarehouseNode extends EntityNode {
     id_field: string
     table_name: string
     timestamp_field: string
+    distinct_id_field: string
 }
 
 export interface ActionsNode extends EntityNode {
diff --git a/frontend/src/scenes/data-warehouse/types.ts b/frontend/src/scenes/data-warehouse/types.ts
index e2ee1440ba3c0..6f5f375d5ee2c 100644
--- a/frontend/src/scenes/data-warehouse/types.ts
+++ b/frontend/src/scenes/data-warehouse/types.ts
@@ -34,6 +34,7 @@ export interface DataWarehouseTableBaseType {
     // used for selecting in trends series
     id_field?: string
     timestamp_field?: string
+    distinct_id_field?: string
 }
 
 export interface DataWarehousePostHogTableType extends DataWarehouseTableBaseType {
diff --git a/frontend/src/scenes/insights/filters/ActionFilter/ActionFilterRow/ActionFilterRow.tsx b/frontend/src/scenes/insights/filters/ActionFilter/ActionFilterRow/ActionFilterRow.tsx
index 22a8d7c00178b..3f6d31c9489b2 100644
--- a/frontend/src/scenes/insights/filters/ActionFilter/ActionFilterRow/ActionFilterRow.tsx
+++ b/frontend/src/scenes/insights/filters/ActionFilter/ActionFilterRow/ActionFilterRow.tsx
@@ -162,8 +162,6 @@ export function ActionFilterRow({
     const { setNodeRef, attributes, transform, transition, listeners, isDragging } = useSortable({ id: filter.uuid })
 
     const propertyFiltersVisible = typeof filter.order === 'number' ? entityFilterVisible[filter.order] : false
-    const mathDisabledReason =
-        filter.type === EntityTypes.DATA_WAREHOUSE ? 'Data Warehouse Series only supports total counts' : ''
 
     let name: string | null | undefined, value: PropertyFilterValue
     const {
@@ -239,6 +237,7 @@ export function ActionFilterRow({
                         name: item?.name ?? '',
                         id_field: item?.id_field,
                         timestamp_field: item?.timestamp_field,
+                        distinct_id_field: item?.distinct_id_field,
                         table_name: item?.name,
                         index,
                     })
@@ -375,7 +374,6 @@ export function ActionFilterRow({
                                         index={index}
                                         onMathSelect={onMathSelect}
                                         disabled={readOnly}
-                                        disabledReason={mathDisabledReason}
                                         style={{ maxWidth: '100%', width: 'initial' }}
                                         mathAvailability={mathAvailability}
                                     />
@@ -385,9 +383,15 @@ export function ActionFilterRow({
                                             <TaxonomicStringPopover
                                                 groupType={TaxonomicFilterGroupType.NumericalEventProperties}
                                                 groupTypes={[
+                                                    TaxonomicFilterGroupType.DataWarehouseProperties,
                                                     TaxonomicFilterGroupType.NumericalEventProperties,
                                                     TaxonomicFilterGroupType.Sessions,
                                                 ]}
+                                                schemaColumns={
+                                                    filter.type == TaxonomicFilterGroupType.DataWarehouse && filter.name
+                                                        ? externalTablesMap[filter.name]?.columns
+                                                        : []
+                                                }
                                                 value={mathProperty}
                                                 onChange={(currentValue) => onMathPropertySelect(index, currentValue)}
                                                 eventNames={name ? [name] : []}
diff --git a/frontend/src/scenes/insights/filters/ActionFilter/entityFilterLogic.ts b/frontend/src/scenes/insights/filters/ActionFilter/entityFilterLogic.ts
index 579916c646f14..c743a8bf4848e 100644
--- a/frontend/src/scenes/insights/filters/ActionFilter/entityFilterLogic.ts
+++ b/frontend/src/scenes/insights/filters/ActionFilter/entityFilterLogic.ts
@@ -23,6 +23,7 @@ export type LocalFilter = ActionFilter & {
     uuid: string
     id_field?: string
     timestamp_field?: string
+    distinct_id_field?: string
     table_name?: string
 }
 
@@ -104,6 +105,7 @@ export const entityFilterLogic = kea<entityFilterLogicType>([
                 index: number
                 id_field?: string
                 timestamp_field?: string
+                distinct_id_field?: string
                 table_name?: string
             }
         ) => ({
@@ -196,7 +198,17 @@ export const entityFilterLogic = kea<entityFilterLogicType>([
         hideModal: () => {
             actions.selectFilter(null)
         },
-        updateFilter: async ({ type, index, name, id, custom_name, id_field, timestamp_field, table_name }) => {
+        updateFilter: async ({
+            type,
+            index,
+            name,
+            id,
+            custom_name,
+            id_field,
+            timestamp_field,
+            distinct_id_field,
+            table_name,
+        }) => {
             actions.setFilters(
                 values.localFilters.map((filter, i) => {
                     if (i === index) {
@@ -210,11 +222,16 @@ export const entityFilterLogic = kea<entityFilterLogicType>([
                                 id_field: typeof id_field === 'undefined' ? filter.id_field : id_field,
                                 timestamp_field:
                                     typeof timestamp_field === 'undefined' ? filter.timestamp_field : timestamp_field,
+                                distinct_id_field:
+                                    typeof distinct_id_field === 'undefined'
+                                        ? filter.distinct_id_field
+                                        : distinct_id_field,
                                 table_name: typeof table_name === 'undefined' ? filter.table_name : table_name,
                             }
                         } else {
                             delete filter.id_field
                             delete filter.timestamp_field
+                            delete filter.distinct_id_field
                             delete filter.table_name
                             return {
                                 ...filter,
diff --git a/frontend/src/types.ts b/frontend/src/types.ts
index b424ac655590f..9a9535ab5d6a0 100644
--- a/frontend/src/types.ts
+++ b/frontend/src/types.ts
@@ -944,6 +944,7 @@ export interface ActionFilter extends EntityFilter {
 export interface DataWarehouseFilter extends ActionFilter {
     id_field: string
     timestamp_field: string
+    distinct_id_field: string
     table_name: string
 }
 
diff --git a/posthog/hogql/database/database.py b/posthog/hogql/database/database.py
index bbf8e8d0cb772..52a65644f76cf 100644
--- a/posthog/hogql/database/database.py
+++ b/posthog/hogql/database/database.py
@@ -199,6 +199,34 @@ def create_hogql_database(
     for table in DataWarehouseTable.objects.filter(team_id=team.pk).exclude(deleted=True):
         tables[table.name] = table.hogql_definition()
 
+    if modifiers.dataWarehouseEventsModifiers:
+        for warehouse_modifier in modifiers.dataWarehouseEventsModifiers:
+            # TODO: add all field mappings
+            if "id" not in tables[warehouse_modifier.table_name].fields.keys():
+                tables[warehouse_modifier.table_name].fields["id"] = ExpressionField(
+                    name="id",
+                    expr=parse_expr(warehouse_modifier.id_field),
+                )
+
+            if "timestamp" not in tables[warehouse_modifier.table_name].fields.keys():
+                tables[warehouse_modifier.table_name].fields["timestamp"] = ExpressionField(
+                    name="timestamp",
+                    expr=ast.Call(name="toDateTime", args=[ast.Field(chain=[warehouse_modifier.timestamp_field])]),
+                )
+
+            # TODO: Need to decide how the distinct_id and person_id fields are going to be handled
+            if "distinct_id" not in tables[warehouse_modifier.table_name].fields.keys():
+                tables[warehouse_modifier.table_name].fields["distinct_id"] = ExpressionField(
+                    name="distinct_id",
+                    expr=parse_expr(warehouse_modifier.distinct_id_field),
+                )
+
+            if "person_id" not in tables[warehouse_modifier.table_name].fields.keys():
+                tables[warehouse_modifier.table_name].fields["person_id"] = ExpressionField(
+                    name="person_id",
+                    expr=parse_expr(warehouse_modifier.distinct_id_field),
+                )
+
     for saved_query in DataWarehouseSavedQuery.objects.filter(team_id=team.pk).exclude(deleted=True):
         tables[saved_query.name] = saved_query.hogql_definition()
 
diff --git a/posthog/hogql_queries/insights/data_warehouse_mixin.py b/posthog/hogql_queries/insights/data_warehouse_mixin.py
new file mode 100644
index 0000000000000..9653915d5e6aa
--- /dev/null
+++ b/posthog/hogql_queries/insights/data_warehouse_mixin.py
@@ -0,0 +1,14 @@
+from posthog.hogql import ast
+from posthog.models.filters.mixins.utils import cached_property
+from posthog.schema import ActionsNode, EventsNode, DataWarehouseNode
+
+
+class DataWarehouseInsightQueryMixin:
+    series: EventsNode | ActionsNode | DataWarehouseNode
+
+    @cached_property
+    def _table_expr(self) -> ast.Field:
+        if isinstance(self.series, DataWarehouseNode):
+            return ast.Field(chain=[self.series.table_name])
+
+        return ast.Field(chain=["events"])
diff --git a/posthog/hogql_queries/insights/trends/aggregation_operations.py b/posthog/hogql_queries/insights/trends/aggregation_operations.py
index 9d60f83f5765d..a0307d625c9a1 100644
--- a/posthog/hogql_queries/insights/trends/aggregation_operations.py
+++ b/posthog/hogql_queries/insights/trends/aggregation_operations.py
@@ -5,6 +5,7 @@
 from posthog.models.team.team import Team
 from posthog.schema import EventsNode, ActionsNode, DataWarehouseNode
 from posthog.models.filters.mixins.utils import cached_property
+from posthog.hogql_queries.insights.data_warehouse_mixin import DataWarehouseInsightQueryMixin
 
 
 class QueryAlternator:
@@ -48,7 +49,7 @@ def replace_select_from(self, join_expr: ast.JoinExpr) -> None:
         self._select_from = join_expr
 
 
-class AggregationOperations:
+class AggregationOperations(DataWarehouseInsightQueryMixin):
     team: Team
     series: Union[EventsNode, ActionsNode, DataWarehouseNode]
     query_date_range: QueryDateRange
@@ -155,6 +156,8 @@ def _math_func(self, method: str, override_chain: Optional[List[str | int]]) ->
 
         if self.series.math_property == "$session_duration":
             chain = ["session_duration"]
+        elif isinstance(self.series, DataWarehouseNode) and self.series.math_property:
+            chain = [self.series.math_property]
         else:
             chain = ["properties", self.series.math_property]
 
@@ -344,6 +347,14 @@ def _events_query(
 
             query = parse_select(
                 """
+                    SELECT
+                        count({id_field}) AS total
+                    FROM {table} AS e
+                    WHERE {events_where_clause}
+                    GROUP BY {person_field}
+                """
+                if isinstance(self.series, DataWarehouseNode)
+                else """
                     SELECT
                         count({id_field}) AS total
                     FROM events AS e
@@ -353,6 +364,7 @@ def _events_query(
                 """,
                 placeholders={
                     "id_field": self._id_field,
+                    "table": self._table_expr,
                     "events_where_clause": where_clause_combined,
                     "sample": sample_value,
                     "person_field": ast.Field(
diff --git a/posthog/hogql_queries/insights/trends/breakdown.py b/posthog/hogql_queries/insights/trends/breakdown.py
index a1576fa2b97f8..45a3a8421e8d8 100644
--- a/posthog/hogql_queries/insights/trends/breakdown.py
+++ b/posthog/hogql_queries/insights/trends/breakdown.py
@@ -241,6 +241,7 @@ def _all_breakdown_values(self) -> List[str | int | float | None]:
                 chart_display_type=self._trends_display().display_type,
                 breakdown_filter=self.query.breakdownFilter,
                 query_date_range=self.query_date_range,
+                modifiers=self.modifiers,
             )
             return cast(List[str | int | float | None], breakdown.get_breakdown_values())
 
diff --git a/posthog/hogql_queries/insights/trends/breakdown_values.py b/posthog/hogql_queries/insights/trends/breakdown_values.py
index 34940650d9dde..7b1522d5f25c5 100644
--- a/posthog/hogql_queries/insights/trends/breakdown_values.py
+++ b/posthog/hogql_queries/insights/trends/breakdown_values.py
@@ -8,7 +8,15 @@
 from posthog.hogql_queries.insights.trends.utils import get_properties_chain
 from posthog.hogql_queries.utils.query_date_range import QueryDateRange
 from posthog.models.team.team import Team
-from posthog.schema import BreakdownFilter, BreakdownType, ChartDisplayType, ActionsNode, EventsNode, DataWarehouseNode
+from posthog.schema import (
+    BreakdownFilter,
+    BreakdownType,
+    ChartDisplayType,
+    ActionsNode,
+    EventsNode,
+    DataWarehouseNode,
+    HogQLQueryModifiers,
+)
 from functools import cached_property
 
 BREAKDOWN_OTHER_STRING_LABEL = "$$_posthog_breakdown_other_$$"
@@ -29,6 +37,7 @@ class BreakdownValues:
     hide_other_aggregation: Optional[bool]
     breakdown_limit: Optional[int]
     query_date_range: QueryDateRange
+    modifiers: HogQLQueryModifiers
 
     def __init__(
         self,
@@ -38,6 +47,7 @@ def __init__(
         chart_display_type: ChartDisplayType,
         breakdown_filter: BreakdownFilter,
         query_date_range: QueryDateRange,
+        modifiers: HogQLQueryModifiers,
     ):
         self.team = team
         self.series = series
@@ -58,6 +68,7 @@ def __init__(
         self.hide_other_aggregation = breakdown_filter.breakdown_hide_other_aggregation
         self.breakdown_limit = breakdown_filter.breakdown_limit
         self.query_date_range = query_date_range
+        self.modifiers = modifiers
 
     def get_breakdown_values(self) -> List[str | int]:
         if self.breakdown_type == "cohort":
@@ -175,6 +186,7 @@ def get_breakdown_values(self) -> List[str | int]:
                 query_type="TrendsQueryBreakdownValues",
                 query=query,
                 team=self.team,
+                modifiers=self.modifiers,
             )
             if response.results and len(response.results) > 0:
                 values = response.results[0][0]
@@ -187,6 +199,7 @@ def get_breakdown_values(self) -> List[str | int]:
                 query_type="TrendsQueryBreakdownValues",
                 query=query,
                 team=self.team,
+                modifiers=self.modifiers,
             )
             value_index = (response.columns or []).index("value")
             values = [row[value_index] for row in response.results or []]
diff --git a/posthog/hogql_queries/insights/trends/data_warehouse_trends_query_builder.py b/posthog/hogql_queries/insights/trends/data_warehouse_trends_query_builder.py
deleted file mode 100644
index ac707c76d7158..0000000000000
--- a/posthog/hogql_queries/insights/trends/data_warehouse_trends_query_builder.py
+++ /dev/null
@@ -1,409 +0,0 @@
-from typing import List, Optional, cast
-from posthog.hogql import ast
-from posthog.hogql.parser import parse_select, parse_expr
-from posthog.hogql.property import property_to_expr
-from posthog.hogql.timings import HogQLTimings
-from posthog.hogql_queries.insights.trends.aggregation_operations import (
-    AggregationOperations,
-)
-from posthog.hogql_queries.insights.trends.breakdown import Breakdown
-from posthog.hogql_queries.insights.trends.display import TrendsDisplay
-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 HogQLQueryModifiers, TrendsQuery, DataWarehouseNode
-from posthog.hogql_queries.insights.trends.trends_query_builder_abstract import TrendsQueryBuilderAbstract
-
-
-class DataWarehouseTrendsQueryBuilder(TrendsQueryBuilderAbstract):
-    query: TrendsQuery
-    team: Team
-    query_date_range: QueryDateRange
-    series: DataWarehouseNode
-    timings: HogQLTimings
-    modifiers: HogQLQueryModifiers
-
-    def __init__(
-        self,
-        trends_query: TrendsQuery,
-        team: Team,
-        query_date_range: QueryDateRange,
-        series: DataWarehouseNode,
-        timings: HogQLTimings,
-        modifiers: HogQLQueryModifiers,
-    ):
-        self.query = trends_query
-        self.team = team
-        self.query_date_range = query_date_range
-        self.series = series
-        self.timings = timings
-        self.modifiers = modifiers
-
-    def build_query(self) -> ast.SelectQuery | ast.SelectUnionQuery:
-        breakdown = self._breakdown(is_actors_query=False)
-
-        events_query: ast.SelectQuery | ast.SelectUnionQuery
-
-        if self._trends_display.should_aggregate_values():
-            events_query = self._get_events_subquery(False, is_actors_query=False, breakdown=breakdown)
-        else:
-            date_subqueries = self._get_date_subqueries(breakdown=breakdown)
-            event_query = self._get_events_subquery(False, is_actors_query=False, breakdown=breakdown)
-
-            events_query = ast.SelectUnionQuery(select_queries=[*date_subqueries, event_query])
-
-        inner_select = self._inner_select_query(inner_query=events_query, breakdown=breakdown)
-        full_query = self._outer_select_query(inner_query=inner_select, breakdown=breakdown)
-
-        return full_query
-
-    def _get_date_subqueries(self, breakdown: Breakdown, ignore_breakdowns: bool = False) -> List[ast.SelectQuery]:
-        if not breakdown.enabled or ignore_breakdowns:
-            return [
-                cast(
-                    ast.SelectQuery,
-                    parse_select(
-                        """
-                        SELECT
-                            0 AS total,
-                            {date_to_start_of_interval} - {number_interval_period} AS day_start
-                        FROM
-                            numbers(
-                                coalesce(dateDiff({interval}, {date_from}, {date_to}), 0)
-                            )
-                    """,
-                        placeholders={
-                            **self.query_date_range.to_placeholders(),
-                        },
-                    ),
-                ),
-                cast(
-                    ast.SelectQuery,
-                    parse_select(
-                        """
-                        SELECT
-                            0 AS total,
-                            {date_from_start_of_interval} AS day_start
-                    """,
-                        placeholders={
-                            **self.query_date_range.to_placeholders(),
-                        },
-                    ),
-                ),
-            ]
-
-        return [
-            cast(
-                ast.SelectQuery,
-                parse_select(
-                    """
-                    SELECT
-                        0 AS total,
-                        ticks.day_start as day_start,
-                        breakdown_value
-                    FROM (
-                        SELECT
-                            {date_to_start_of_interval} - {number_interval_period} AS day_start
-                        FROM
-                            numbers(
-                                coalesce(dateDiff({interval}, {date_from}, {date_to}), 0)
-                            )
-                        UNION ALL
-                        SELECT {date_from_start_of_interval} AS day_start
-                    ) as ticks
-                    CROSS JOIN (
-                        SELECT breakdown_value
-                        FROM (
-                            SELECT {cross_join_breakdown_values}
-                        )
-                        ARRAY JOIN breakdown_value as breakdown_value
-                    ) as sec
-                    ORDER BY breakdown_value, day_start
-                """,
-                    placeholders={
-                        **self.query_date_range.to_placeholders(),
-                        **breakdown.placeholders(),
-                    },
-                ),
-            )
-        ]
-
-    def _get_events_subquery(
-        self,
-        no_modifications: Optional[bool],
-        is_actors_query: bool,
-        breakdown: Breakdown,
-        breakdown_values_override: Optional[str | int] = None,
-        actors_query_time_frame: Optional[str | int] = None,
-    ) -> ast.SelectQuery:
-        day_start = ast.Alias(
-            alias="day_start",
-            expr=ast.Call(
-                name=f"toStartOf{self.query_date_range.interval_name.title()}",
-                args=[ast.Call(name="toDateTime", args=[ast.Field(chain=[self.series.timestamp_field])])],
-            ),
-        )
-
-        events_filter = self._events_filter(
-            ignore_breakdowns=False,
-            breakdown=breakdown,
-            is_actors_query=is_actors_query,
-        )
-
-        default_query = cast(
-            ast.SelectQuery,
-            parse_select(
-                """
-                SELECT
-                    {aggregation_operation} AS total
-                FROM {table} AS e
-                WHERE {events_filter}
-            """,
-                placeholders={
-                    "events_filter": events_filter,
-                    "aggregation_operation": self._aggregation_operation.select_aggregation(),
-                    "table": self._table_expr,
-                },
-            ),
-        )
-
-        default_query.group_by = []
-
-        if not self._trends_display.should_aggregate_values() and not is_actors_query:
-            default_query.select.append(day_start)
-            default_query.group_by.append(ast.Field(chain=["day_start"]))
-
-        # TODO: Move this logic into the below branches when working on adding breakdown support for the person modal
-        if is_actors_query:
-            default_query.select = [ast.Alias(alias="person_id", expr=ast.Field(chain=["e", "person_id"]))]
-            default_query.distinct = True
-            default_query.group_by = []
-
-        # No breakdowns and no complex series aggregation
-        if (
-            not breakdown.enabled
-            and not self._aggregation_operation.requires_query_orchestration()
-            and not self._aggregation_operation.aggregating_on_session_duration()
-        ) or no_modifications is True:
-            return default_query
-        # Both breakdowns and complex series aggregation
-        elif breakdown.enabled and self._aggregation_operation.requires_query_orchestration():
-            raise NotImplementedError(
-                "Breakdowns and complex series aggregation are not supported for Data Warehouse queries"
-            )
-        # Breakdowns and session duration math property
-        elif breakdown.enabled and self._aggregation_operation.aggregating_on_session_duration():
-            raise NotImplementedError(
-                "Breakdowns and session duration math property are not supported for Data Warehouse queries"
-            )
-        # Just breakdowns
-        elif breakdown.enabled:
-            if not is_actors_query:
-                default_query.select.append(breakdown.column_expr())
-                default_query.group_by.append(ast.Field(chain=["breakdown_value"]))
-        # Just session duration math property
-        elif self._aggregation_operation.aggregating_on_session_duration():
-            raise NotImplementedError("Session duration math property is not supported for Data Warehouse queries")
-        # Just complex series aggregation
-        elif self._aggregation_operation.requires_query_orchestration():
-            raise NotImplementedError("Complex series aggregation is not supported for Data Warehouse queries")
-
-        return default_query
-
-    def _outer_select_query(self, breakdown: Breakdown, inner_query: ast.SelectQuery) -> ast.SelectQuery:
-        query = cast(
-            ast.SelectQuery,
-            parse_select(
-                """
-                SELECT
-                    groupArray(day_start) AS date,
-                    groupArray(count) AS total
-                FROM {inner_query}
-            """,
-                placeholders={"inner_query": inner_query},
-            ),
-        )
-
-        query = self._trends_display.modify_outer_query(
-            outer_query=query,
-            inner_query=inner_query,
-            dates_queries=ast.SelectUnionQuery(
-                select_queries=self._get_date_subqueries(ignore_breakdowns=True, breakdown=breakdown)
-            ),
-        )
-
-        query.order_by = [ast.OrderExpr(expr=ast.Call(name="sum", args=[ast.Field(chain=["count"])]), order="DESC")]
-
-        if breakdown.enabled:
-            query.select.append(
-                ast.Alias(
-                    alias="breakdown_value",
-                    expr=ast.Call(
-                        name="ifNull",
-                        args=[
-                            ast.Call(name="toString", args=[ast.Field(chain=["breakdown_value"])]),
-                            ast.Constant(value=""),
-                        ],
-                    ),
-                )
-            )
-            query.group_by = [ast.Field(chain=["breakdown_value"])]
-            query.order_by.append(ast.OrderExpr(expr=ast.Field(chain=["breakdown_value"]), order="ASC"))
-
-        return query
-
-    def _inner_select_query(
-        self, breakdown: Breakdown, inner_query: ast.SelectQuery | ast.SelectUnionQuery
-    ) -> ast.SelectQuery:
-        query = cast(
-            ast.SelectQuery,
-            parse_select(
-                """
-                SELECT
-                    sum(total) AS count
-                FROM {inner_query}
-            """,
-                placeholders={"inner_query": inner_query},
-            ),
-        )
-
-        if (
-            self.query.trendsFilter is not None
-            and self.query.trendsFilter.smoothingIntervals is not None
-            and self.query.trendsFilter.smoothingIntervals > 1
-        ):
-            rolling_average = ast.Alias(
-                alias="count",
-                expr=ast.Call(
-                    name="floor",
-                    args=[
-                        ast.WindowFunction(
-                            name="avg",
-                            args=[ast.Call(name="sum", args=[ast.Field(chain=["total"])])],
-                            over_expr=ast.WindowExpr(
-                                order_by=[ast.OrderExpr(expr=ast.Field(chain=["day_start"]), order="ASC")],
-                                frame_method="ROWS",
-                                frame_start=ast.WindowFrameExpr(
-                                    frame_type="PRECEDING",
-                                    frame_value=int(self.query.trendsFilter.smoothingIntervals - 1),
-                                ),
-                                frame_end=ast.WindowFrameExpr(frame_type="CURRENT ROW"),
-                            ),
-                        )
-                    ],
-                ),
-            )
-            query.select = [rolling_average]
-
-        query.group_by = []
-        query.order_by = []
-
-        if not self._trends_display.should_aggregate_values():
-            query.select.append(ast.Field(chain=["day_start"]))
-            query.group_by.append(ast.Field(chain=["day_start"]))
-            query.order_by.append(ast.OrderExpr(expr=ast.Field(chain=["day_start"]), order="ASC"))
-
-        if breakdown.enabled:
-            query.select.append(ast.Field(chain=["breakdown_value"]))
-            query.group_by.append(ast.Field(chain=["breakdown_value"]))
-            query.order_by.append(ast.OrderExpr(expr=ast.Field(chain=["breakdown_value"]), order="ASC"))
-
-        if self._trends_display.should_wrap_inner_query():
-            query = self._trends_display.wrap_inner_query(query, breakdown.enabled)
-            if breakdown.enabled:
-                query.select.append(ast.Field(chain=["breakdown_value"]))
-
-        return query
-
-    def _events_filter(
-        self,
-        is_actors_query: bool,
-        breakdown: Breakdown | None,
-        ignore_breakdowns: bool = False,
-        breakdown_values_override: Optional[str | int] = None,
-        actors_query_time_frame: Optional[str | int] = None,
-    ) -> ast.Expr:
-        series = self.series
-        filters: List[ast.Expr] = []
-
-        filters.extend(
-            [
-                parse_expr(
-                    "{timestamp_field} >= {date_from_with_adjusted_start_of_interval}",
-                    placeholders={
-                        "timestamp_field": ast.Call(
-                            name="toDateTime", args=[ast.Field(chain=[self.series.timestamp_field])]
-                        ),
-                        **self.query_date_range.to_placeholders(),
-                    },
-                ),
-                parse_expr(
-                    "{timestamp_field} <= {date_to}",
-                    placeholders={
-                        "timestamp_field": ast.Call(
-                            name="toDateTime", args=[ast.Field(chain=[self.series.timestamp_field])]
-                        ),
-                        **self.query_date_range.to_placeholders(),
-                    },
-                ),
-            ]
-        )
-
-        # Properties
-        if self.query.properties is not None and self.query.properties != []:
-            filters.append(property_to_expr(self.query.properties, self.team))
-
-        # Series Filters
-        if series.properties is not None and series.properties != []:
-            filters.append(property_to_expr(series.properties, self.team))
-
-        # Breakdown
-        if not ignore_breakdowns and breakdown is not None:
-            if breakdown.enabled and not breakdown.is_histogram_breakdown:
-                breakdown_filter = breakdown.events_where_filter()
-                if breakdown_filter is not None:
-                    filters.append(breakdown_filter)
-
-        if len(filters) == 0:
-            return ast.Constant(value=True)
-
-        return ast.And(exprs=filters)
-
-    def _breakdown(self, is_actors_query: bool, breakdown_values_override: Optional[str | int] = None):
-        return Breakdown(
-            team=self.team,
-            query=self.query,
-            series=self.series,
-            query_date_range=self.query_date_range,
-            timings=self.timings,
-            modifiers=self.modifiers,
-            events_filter=self._events_filter(
-                breakdown=None,  # Passing in None because we know we dont actually need it
-                ignore_breakdowns=True,
-                is_actors_query=is_actors_query,
-                breakdown_values_override=breakdown_values_override,
-            ),
-            breakdown_values_override=[breakdown_values_override] if breakdown_values_override is not None else None,
-        )
-
-    @cached_property
-    def _aggregation_operation(self) -> AggregationOperations:
-        if self.series.math is not None and self.series.math != "total":
-            raise NotImplementedError("Math types other than total are not supported for Data Warehouse queries")
-
-        return AggregationOperations(
-            self.team, self.series, self.query_date_range, self._trends_display.should_aggregate_values()
-        )
-
-    @cached_property
-    def _trends_display(self) -> TrendsDisplay:
-        display = (
-            self.query.trendsFilter.display
-            if self.query.trendsFilter is not None and self.query.trendsFilter.display is not None
-            else None
-        )
-        return TrendsDisplay(display)
-
-    @cached_property
-    def _table_expr(self) -> ast.Field:
-        return ast.Field(chain=[self.series.table_name])
diff --git a/posthog/hogql_queries/insights/trends/test/__snapshots__/test_data_warehouse_query_builder.ambr b/posthog/hogql_queries/insights/trends/test/__snapshots__/test_data_warehouse_query_builder.ambr
index aef6ad4db2ffd..47eda0a092ab7 100644
--- a/posthog/hogql_queries/insights/trends/test/__snapshots__/test_data_warehouse_query_builder.ambr
+++ b/posthog/hogql_queries/insights/trends/test/__snapshots__/test_data_warehouse_query_builder.ambr
@@ -157,32 +157,6 @@
                      allow_experimental_object_type=1
   '''
 # ---
-# name: TestDataWarehouseQueryBuilder.test_trends_other_property_invalid
-  '''
-  SELECT groupArray(day_start) AS date,
-         groupArray(count) AS total
-  FROM
-    (SELECT sum(total) AS count,
-            day_start AS day_start
-     FROM
-       (SELECT 0 AS total,
-               minus(toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), toIntervalDay(numbers.number)) AS day_start
-        FROM numbers(coalesce(dateDiff('day', assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)) AS numbers
-        UNION ALL SELECT 0 AS total,
-                         toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC'))) AS day_start
-        UNION ALL SELECT count(e.id) AS total,
-                         toStartOfDay(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC')) AS day_start
-        FROM s3('http://host.docker.internal:19000/posthog/test_storage_bucket-posthog.hogql.datawarehouse.trendquery/*.parquet', 'object_storage_root_user', 'object_storage_root_password', 'Parquet', 'id String, prop_1 String, prop_2 String, created DateTime64(3, \'UTC\')') AS e
-        WHERE and(ifNull(greaterOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0), true)
-        GROUP BY day_start)
-     GROUP BY day_start
-     ORDER BY day_start ASC)
-  ORDER BY sum(count) DESC
-  LIMIT 100 SETTINGS readonly=2,
-                     max_execution_time=60,
-                     allow_experimental_object_type=1
-  '''
-# ---
 # name: TestDataWarehouseQueryBuilder.test_trends_property
   '''
   SELECT groupArray(day_start) AS date,
diff --git a/posthog/hogql_queries/insights/trends/test/__snapshots__/test_trends_data_warehouse_query.ambr b/posthog/hogql_queries/insights/trends/test/__snapshots__/test_trends_data_warehouse_query.ambr
new file mode 100644
index 0000000000000..db9e8e1d45000
--- /dev/null
+++ b/posthog/hogql_queries/insights/trends/test/__snapshots__/test_trends_data_warehouse_query.ambr
@@ -0,0 +1,185 @@
+# serializer version: 1
+# name: TestTrendsDataWarehouseQuery.test_trends_breakdown
+  '''
+  SELECT e.prop_1 AS value,
+         count(e.id) AS count
+  FROM s3('http://host.docker.internal:19000/posthog/test_storage_bucket-posthog.hogql.datawarehouse.trendquery/*.parquet', 'object_storage_root_user', 'object_storage_root_password', 'Parquet', 'id String, prop_1 String, prop_2 String, created DateTime64(3, \'UTC\')') AS e
+  WHERE and(and(ifNull(greaterOrEquals(toTimeZone(e.created, 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toTimeZone(e.created, 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)), and(ifNull(greaterOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)))
+  GROUP BY value
+  ORDER BY count DESC, value DESC
+  LIMIT 26 SETTINGS readonly=2,
+                    max_execution_time=60,
+                    allow_experimental_object_type=1
+  '''
+# ---
+# name: TestTrendsDataWarehouseQuery.test_trends_breakdown.1
+  '''
+  SELECT groupArray(day_start) AS date,
+         groupArray(count) AS total,
+         ifNull(toString(breakdown_value), '') AS breakdown_value
+  FROM
+    (SELECT sum(total) AS count,
+            day_start AS day_start,
+            breakdown_value AS breakdown_value
+     FROM
+       (SELECT 0 AS total,
+               ticks.day_start AS day_start,
+               sec.breakdown_value AS breakdown_value
+        FROM
+          (SELECT minus(toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), toIntervalDay(numbers.number)) AS day_start
+           FROM numbers(coalesce(dateDiff('day', assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)) AS numbers
+           UNION ALL SELECT toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC'))) AS day_start) AS ticks
+        CROSS JOIN
+          (SELECT breakdown_value
+           FROM
+             (SELECT ['d', 'c', 'b', 'a'] AS breakdown_value) ARRAY
+           JOIN breakdown_value AS breakdown_value) AS sec
+        ORDER BY sec.breakdown_value ASC, day_start ASC
+        UNION ALL SELECT count(e.id) AS total,
+                         toStartOfDay(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC')) AS day_start,
+                         transform(ifNull(e.prop_1, '$$_posthog_breakdown_null_$$'), ['d', 'c', 'b', 'a'], ['d', 'c', 'b', 'a'], '$$_posthog_breakdown_other_$$') AS breakdown_value
+        FROM s3('http://host.docker.internal:19000/posthog/test_storage_bucket-posthog.hogql.datawarehouse.trendquery/*.parquet', 'object_storage_root_user', 'object_storage_root_password', 'Parquet', 'id String, prop_1 String, prop_2 String, created DateTime64(3, \'UTC\')') AS e
+        WHERE and(ifNull(greaterOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0), or(equals(e.prop_1, 'd'), equals(e.prop_1, 'c'), equals(e.prop_1, 'b'), equals(e.prop_1, 'a')))
+        GROUP BY day_start,
+                 breakdown_value)
+     GROUP BY day_start,
+              breakdown_value
+     ORDER BY day_start ASC, breakdown_value ASC)
+  GROUP BY breakdown_value
+  ORDER BY sum(count) DESC, breakdown_value ASC
+  LIMIT 100 SETTINGS readonly=2,
+                     max_execution_time=60,
+                     allow_experimental_object_type=1
+  '''
+# ---
+# name: TestTrendsDataWarehouseQuery.test_trends_breakdown_with_property
+  '''
+  SELECT e.prop_1 AS value,
+         count(e.id) AS count
+  FROM s3('http://host.docker.internal:19000/posthog/test_storage_bucket-posthog.hogql.datawarehouse.trendquery/*.parquet', 'object_storage_root_user', 'object_storage_root_password', 'Parquet', 'id String, prop_1 String, prop_2 String, created DateTime64(3, \'UTC\')') AS e
+  WHERE and(and(ifNull(greaterOrEquals(toTimeZone(e.created, 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toTimeZone(e.created, 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)), and(ifNull(greaterOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0), equals(e.prop_1, 'a')))
+  GROUP BY value
+  ORDER BY count DESC, value DESC
+  LIMIT 26 SETTINGS readonly=2,
+                    max_execution_time=60,
+                    allow_experimental_object_type=1
+  '''
+# ---
+# name: TestTrendsDataWarehouseQuery.test_trends_breakdown_with_property.1
+  '''
+  SELECT groupArray(day_start) AS date,
+         groupArray(count) AS total,
+         ifNull(toString(breakdown_value), '') AS breakdown_value
+  FROM
+    (SELECT sum(total) AS count,
+            day_start AS day_start,
+            breakdown_value AS breakdown_value
+     FROM
+       (SELECT 0 AS total,
+               ticks.day_start AS day_start,
+               sec.breakdown_value AS breakdown_value
+        FROM
+          (SELECT minus(toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), toIntervalDay(numbers.number)) AS day_start
+           FROM numbers(coalesce(dateDiff('day', assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)) AS numbers
+           UNION ALL SELECT toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC'))) AS day_start) AS ticks
+        CROSS JOIN
+          (SELECT breakdown_value
+           FROM
+             (SELECT ['a'] AS breakdown_value) ARRAY
+           JOIN breakdown_value AS breakdown_value) AS sec
+        ORDER BY sec.breakdown_value ASC, day_start ASC
+        UNION ALL SELECT count(e.id) AS total,
+                         toStartOfDay(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC')) AS day_start,
+                         transform(ifNull(e.prop_1, '$$_posthog_breakdown_null_$$'), ['a'], ['a'], '$$_posthog_breakdown_other_$$') AS breakdown_value
+        FROM s3('http://host.docker.internal:19000/posthog/test_storage_bucket-posthog.hogql.datawarehouse.trendquery/*.parquet', 'object_storage_root_user', 'object_storage_root_password', 'Parquet', 'id String, prop_1 String, prop_2 String, created DateTime64(3, \'UTC\')') AS e
+        WHERE and(ifNull(greaterOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0), equals(e.prop_1, 'a'), equals(e.prop_1, 'a'))
+        GROUP BY day_start,
+                 breakdown_value)
+     GROUP BY day_start,
+              breakdown_value
+     ORDER BY day_start ASC, breakdown_value ASC)
+  GROUP BY breakdown_value
+  ORDER BY sum(count) DESC, breakdown_value ASC
+  LIMIT 100 SETTINGS readonly=2,
+                     max_execution_time=60,
+                     allow_experimental_object_type=1
+  '''
+# ---
+# name: TestTrendsDataWarehouseQuery.test_trends_data_warehouse
+  '''
+  SELECT groupArray(day_start) AS date,
+         groupArray(count) AS total
+  FROM
+    (SELECT sum(total) AS count,
+            day_start AS day_start
+     FROM
+       (SELECT 0 AS total,
+               minus(toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), toIntervalDay(numbers.number)) AS day_start
+        FROM numbers(coalesce(dateDiff('day', assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)) AS numbers
+        UNION ALL SELECT 0 AS total,
+                         toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC'))) AS day_start
+        UNION ALL SELECT count(e.id) AS total,
+                         toStartOfDay(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC')) AS day_start
+        FROM s3('http://host.docker.internal:19000/posthog/test_storage_bucket-posthog.hogql.datawarehouse.trendquery/*.parquet', 'object_storage_root_user', 'object_storage_root_password', 'Parquet', 'id String, prop_1 String, prop_2 String, created DateTime64(3, \'UTC\')') AS e
+        WHERE and(ifNull(greaterOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0))
+        GROUP BY day_start)
+     GROUP BY day_start
+     ORDER BY day_start ASC)
+  ORDER BY sum(count) DESC
+  LIMIT 100 SETTINGS readonly=2,
+                     max_execution_time=60,
+                     allow_experimental_object_type=1
+  '''
+# ---
+# name: TestTrendsDataWarehouseQuery.test_trends_entity_property
+  '''
+  SELECT groupArray(day_start) AS date,
+         groupArray(count) AS total
+  FROM
+    (SELECT sum(total) AS count,
+            day_start AS day_start
+     FROM
+       (SELECT 0 AS total,
+               minus(toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), toIntervalDay(numbers.number)) AS day_start
+        FROM numbers(coalesce(dateDiff('day', assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)) AS numbers
+        UNION ALL SELECT 0 AS total,
+                         toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC'))) AS day_start
+        UNION ALL SELECT count(e.id) AS total,
+                         toStartOfDay(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC')) AS day_start
+        FROM s3('http://host.docker.internal:19000/posthog/test_storage_bucket-posthog.hogql.datawarehouse.trendquery/*.parquet', 'object_storage_root_user', 'object_storage_root_password', 'Parquet', 'id String, prop_1 String, prop_2 String, created DateTime64(3, \'UTC\')') AS e
+        WHERE and(ifNull(greaterOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0), equals(e.prop_1, 'a'))
+        GROUP BY day_start)
+     GROUP BY day_start
+     ORDER BY day_start ASC)
+  ORDER BY sum(count) DESC
+  LIMIT 100 SETTINGS readonly=2,
+                     max_execution_time=60,
+                     allow_experimental_object_type=1
+  '''
+# ---
+# name: TestTrendsDataWarehouseQuery.test_trends_property
+  '''
+  SELECT groupArray(day_start) AS date,
+         groupArray(count) AS total
+  FROM
+    (SELECT sum(total) AS count,
+            day_start AS day_start
+     FROM
+       (SELECT 0 AS total,
+               minus(toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), toIntervalDay(numbers.number)) AS day_start
+        FROM numbers(coalesce(dateDiff('day', assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0)) AS numbers
+        UNION ALL SELECT 0 AS total,
+                         toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC'))) AS day_start
+        UNION ALL SELECT count(e.id) AS total,
+                         toStartOfDay(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC')) AS day_start
+        FROM s3('http://host.docker.internal:19000/posthog/test_storage_bucket-posthog.hogql.datawarehouse.trendquery/*.parquet', 'object_storage_root_user', 'object_storage_root_password', 'Parquet', 'id String, prop_1 String, prop_2 String, created DateTime64(3, \'UTC\')') AS e
+        WHERE and(ifNull(greaterOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), toStartOfDay(assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-01 00:00:00', 6, 'UTC')))), 0), ifNull(lessOrEquals(toDateTime(toTimeZone(e.created, 'UTC'), 'UTC'), assumeNotNull(parseDateTime64BestEffortOrNull('2023-01-07 23:59:59', 6, 'UTC'))), 0), equals(e.prop_1, 'a'))
+        GROUP BY day_start)
+     GROUP BY day_start
+     ORDER BY day_start ASC)
+  ORDER BY sum(count) DESC
+  LIMIT 100 SETTINGS readonly=2,
+                     max_execution_time=60,
+                     allow_experimental_object_type=1
+  '''
+# ---
diff --git a/posthog/hogql_queries/insights/trends/test/test_data_warehouse_query_builder.py b/posthog/hogql_queries/insights/trends/test/test_trends_data_warehouse_query.py
similarity index 81%
rename from posthog/hogql_queries/insights/trends/test/test_data_warehouse_query_builder.py
rename to posthog/hogql_queries/insights/trends/test/test_trends_data_warehouse_query.py
index 96458a0645de0..e47de87fbada1 100644
--- a/posthog/hogql_queries/insights/trends/test/test_data_warehouse_query_builder.py
+++ b/posthog/hogql_queries/insights/trends/test/test_trends_data_warehouse_query.py
@@ -4,7 +4,7 @@
 
 from posthog.hogql.query import execute_hogql_query
 from posthog.hogql.timings import HogQLTimings
-from posthog.hogql_queries.insights.trends.data_warehouse_trends_query_builder import DataWarehouseTrendsQueryBuilder
+from posthog.hogql_queries.insights.trends.trends_query_builder import TrendsQueryBuilder
 from posthog.hogql_queries.utils.query_date_range import QueryDateRange
 from posthog.schema import (
     BreakdownFilter,
@@ -12,6 +12,7 @@
     ChartDisplayType,
     DateRange,
     DataWarehouseNode,
+    DataWarehouseEventsModifier,
     TrendsQuery,
     TrendsFilter,
 )
@@ -41,7 +42,7 @@
 TEST_BUCKET = "test_storage_bucket-posthog.hogql.datawarehouse.trendquery"
 
 
-class TestDataWarehouseQueryBuilder(ClickhouseTestMixin, BaseTest):
+class TestTrendsDataWarehouseQuery(ClickhouseTestMixin, BaseTest):
     def teardown_method(self, method) -> None:
         s3 = resource(
             "s3",
@@ -66,7 +67,16 @@ def get_response(self, trends_query: TrendsQuery):
         modifiers = create_default_modifiers_for_team(self.team)
 
         if isinstance(trends_query.series[0], DataWarehouseNode):
-            query_builder = DataWarehouseTrendsQueryBuilder(
+            series = trends_query.series[0]
+            modifiers.dataWarehouseEventsModifiers = [
+                DataWarehouseEventsModifier(
+                    table_name=series.table_name,
+                    timestamp_field=series.timestamp_field,
+                    id_field=series.id_field,
+                    distinct_id_field=series.distinct_id_field,
+                )
+            ]
+            query_builder = TrendsQueryBuilder(
                 trends_query=trends_query,
                 team=self.team,
                 query_date_range=query_date_range,
@@ -84,6 +94,7 @@ def get_response(self, trends_query: TrendsQuery):
             query=query,
             team=self.team,
             timings=timings,
+            modifiers=modifiers,
         )
 
     def create_parquet_file(self):
@@ -143,7 +154,15 @@ def test_trends_data_warehouse(self):
         trends_query = TrendsQuery(
             kind="TrendsQuery",
             dateRange=DateRange(date_from="2023-01-01"),
-            series=[DataWarehouseNode(id=table_name, table_name=table_name, id_field="id", timestamp_field="created")],
+            series=[
+                DataWarehouseNode(
+                    id=table_name,
+                    table_name=table_name,
+                    id_field="id",
+                    distinct_id_field="customer_email",
+                    timestamp_field="created",
+                )
+            ],
         )
 
         with freeze_time("2023-01-07"):
@@ -166,6 +185,7 @@ def test_trends_entity_property(self):
                     table_name=table_name,
                     id_field="id",
                     timestamp_field="created",
+                    distinct_id_field="customer_email",
                     properties=clean_entity_properties([{"key": "prop_1", "value": "a", "type": "data_warehouse"}]),
                 )
             ],
@@ -185,7 +205,15 @@ def test_trends_property(self):
         trends_query = TrendsQuery(
             kind="TrendsQuery",
             dateRange=DateRange(date_from="2023-01-01"),
-            series=[DataWarehouseNode(id=table_name, table_name=table_name, id_field="id", timestamp_field="created")],
+            series=[
+                DataWarehouseNode(
+                    id=table_name,
+                    table_name=table_name,
+                    id_field="id",
+                    distinct_id_field="customer_email",
+                    timestamp_field="created",
+                )
+            ],
             properties=clean_entity_properties([{"key": "prop_1", "value": "a", "type": "data_warehouse"}]),
         )
 
@@ -203,7 +231,15 @@ def test_trends_breakdown(self):
         trends_query = TrendsQuery(
             kind="TrendsQuery",
             dateRange=DateRange(date_from="2023-01-01"),
-            series=[DataWarehouseNode(id=table_name, table_name=table_name, id_field="id", timestamp_field="created")],
+            series=[
+                DataWarehouseNode(
+                    id=table_name,
+                    table_name=table_name,
+                    id_field="id",
+                    distinct_id_field="customer_email",
+                    timestamp_field="created",
+                )
+            ],
             breakdownFilter=BreakdownFilter(breakdown_type=BreakdownType.data_warehouse, breakdown="prop_1"),
         )
 
@@ -232,7 +268,15 @@ def test_trends_breakdown_with_property(self):
         trends_query = TrendsQuery(
             kind="TrendsQuery",
             dateRange=DateRange(date_from="2023-01-01"),
-            series=[DataWarehouseNode(id=table_name, table_name=table_name, id_field="id", timestamp_field="created")],
+            series=[
+                DataWarehouseNode(
+                    id=table_name,
+                    table_name=table_name,
+                    id_field="id",
+                    distinct_id_field="customer_email",
+                    timestamp_field="created",
+                )
+            ],
             properties=clean_entity_properties([{"key": "prop_1", "value": "a", "type": "data_warehouse"}]),
             breakdownFilter=BreakdownFilter(breakdown_type=BreakdownType.data_warehouse, breakdown="prop_1"),
         )
@@ -253,7 +297,15 @@ def assert_column_names_with_display_type(self, display_type: ChartDisplayType):
         trends_query = TrendsQuery(
             kind="TrendsQuery",
             dateRange=DateRange(date_from="2023-01-01"),
-            series=[DataWarehouseNode(id=table_name, table_name=table_name, id_field="id", timestamp_field="created")],
+            series=[
+                DataWarehouseNode(
+                    id=table_name,
+                    table_name=table_name,
+                    id_field="id",
+                    distinct_id_field="customer_email",
+                    timestamp_field="created",
+                )
+            ],
             trendsFilter=TrendsFilter(display=display_type),
         )
 
diff --git a/posthog/hogql_queries/insights/trends/trends_query_builder.py b/posthog/hogql_queries/insights/trends/trends_query_builder.py
index 0f72802f10405..ed5d867b48b75 100644
--- a/posthog/hogql_queries/insights/trends/trends_query_builder.py
+++ b/posthog/hogql_queries/insights/trends/trends_query_builder.py
@@ -3,6 +3,7 @@
 from posthog.hogql.parser import parse_expr, parse_select
 from posthog.hogql.property import action_to_expr, property_to_expr
 from posthog.hogql.timings import HogQLTimings
+from posthog.hogql_queries.insights.data_warehouse_mixin import DataWarehouseInsightQueryMixin
 from posthog.hogql_queries.insights.trends.aggregation_operations import (
     AggregationOperations,
 )
@@ -13,15 +14,21 @@
 from posthog.models.action.action import Action
 from posthog.models.filters.mixins.utils import cached_property
 from posthog.models.team.team import Team
-from posthog.schema import ActionsNode, EventsNode, HogQLQueryModifiers, TrendsQuery, ChartDisplayType
-from posthog.hogql_queries.insights.trends.trends_query_builder_abstract import TrendsQueryBuilderAbstract
+from posthog.schema import (
+    ActionsNode,
+    DataWarehouseNode,
+    EventsNode,
+    HogQLQueryModifiers,
+    TrendsQuery,
+    ChartDisplayType,
+)
 
 
-class TrendsQueryBuilder(TrendsQueryBuilderAbstract):
+class TrendsQueryBuilder(DataWarehouseInsightQueryMixin):
     query: TrendsQuery
     team: Team
     query_date_range: QueryDateRange
-    series: EventsNode | ActionsNode
+    series: EventsNode | ActionsNode | DataWarehouseNode
     timings: HogQLTimings
     modifiers: HogQLQueryModifiers
 
@@ -30,7 +37,7 @@ def __init__(
         trends_query: TrendsQuery,
         team: Team,
         query_date_range: QueryDateRange,
-        series: EventsNode | ActionsNode,
+        series: EventsNode | ActionsNode | DataWarehouseNode,
         timings: HogQLTimings,
         modifiers: HogQLQueryModifiers,
     ):
@@ -179,13 +186,21 @@ def _get_events_subquery(
             ast.SelectQuery,
             parse_select(
                 """
-                SELECT
-                    {aggregation_operation} AS total
-                FROM events AS e
-                SAMPLE {sample}
-                WHERE {events_filter}
-            """,
+                    SELECT
+                        {aggregation_operation} AS total
+                    FROM {table} AS e
+                    WHERE {events_filter}
+                """
+                if isinstance(self.series, DataWarehouseNode)
+                else """
+                    SELECT
+                        {aggregation_operation} AS total
+                    FROM {table} AS e
+                    SAMPLE {sample}
+                    WHERE {events_filter}
+                """,
                 placeholders={
+                    "table": self._table_expr,
                     "events_filter": events_filter,
                     "aggregation_operation": self._aggregation_operation.select_aggregation(),
                     "sample": self._sample_value(),
diff --git a/posthog/hogql_queries/insights/trends/trends_query_builder_abstract.py b/posthog/hogql_queries/insights/trends/trends_query_builder_abstract.py
deleted file mode 100644
index d321fc0a77ecf..0000000000000
--- a/posthog/hogql_queries/insights/trends/trends_query_builder_abstract.py
+++ /dev/null
@@ -1,52 +0,0 @@
-import abc
-from posthog.hogql import ast
-from typing import List, Optional
-from posthog.hogql_queries.insights.trends.breakdown import Breakdown
-
-
-class TrendsQueryBuilderAbstract(metaclass=abc.ABCMeta):
-    @abc.abstractmethod
-    def build_query(self) -> ast.SelectQuery | ast.SelectUnionQuery:
-        pass
-
-    # Private functions not really necessary but keeping here for uniformity for now
-
-    @abc.abstractmethod
-    def _get_date_subqueries(self, breakdown: Breakdown, ignore_breakdowns: bool = False) -> List[ast.SelectQuery]:
-        pass
-
-    @abc.abstractmethod
-    def _get_events_subquery(
-        self,
-        no_modifications: Optional[bool],
-        is_actors_query: bool,
-        breakdown: Breakdown,
-        breakdown_values_override: Optional[str | int] = None,
-        actors_query_time_frame: Optional[str | int] = None,
-    ) -> ast.SelectQuery:
-        pass
-
-    @abc.abstractmethod
-    def _outer_select_query(self, breakdown: Breakdown, inner_query: ast.SelectQuery) -> ast.SelectQuery:
-        pass
-
-    @abc.abstractmethod
-    def _inner_select_query(
-        self, breakdown: Breakdown, inner_query: ast.SelectQuery | ast.SelectUnionQuery
-    ) -> ast.SelectQuery:
-        pass
-
-    @abc.abstractmethod
-    def _events_filter(
-        self,
-        is_actors_query: bool,
-        breakdown: Breakdown | None,
-        ignore_breakdowns: bool = False,
-        breakdown_values_override: Optional[str | int] = None,
-        actors_query_time_frame: Optional[str | int] = None,
-    ) -> ast.Expr:
-        pass
-
-    @abc.abstractmethod
-    def _breakdown(self, is_actors_query: bool, breakdown_values_override: Optional[str | int] = None):
-        pass
diff --git a/posthog/hogql_queries/insights/trends/trends_query_runner.py b/posthog/hogql_queries/insights/trends/trends_query_runner.py
index ac741cc45f729..6cf84dcb5357b 100644
--- a/posthog/hogql_queries/insights/trends/trends_query_runner.py
+++ b/posthog/hogql_queries/insights/trends/trends_query_runner.py
@@ -29,9 +29,7 @@
     BREAKDOWN_OTHER_STRING_LABEL,
 )
 from posthog.hogql_queries.insights.trends.display import TrendsDisplay
-from posthog.hogql_queries.insights.trends.trends_query_builder_abstract import TrendsQueryBuilderAbstract
 from posthog.hogql_queries.insights.trends.trends_query_builder import TrendsQueryBuilder
-from posthog.hogql_queries.insights.trends.data_warehouse_trends_query_builder import DataWarehouseTrendsQueryBuilder
 from posthog.hogql_queries.insights.trends.series_with_extras import SeriesWithExtras
 from posthog.hogql_queries.query_runner import QueryRunner
 from posthog.hogql_queries.utils.formula_ast import FormulaAST
@@ -61,6 +59,7 @@
     TrendsQuery,
     TrendsQueryResponse,
     HogQLQueryModifiers,
+    DataWarehouseEventsModifier,
 )
 from posthog.utils import format_label_date
 
@@ -122,26 +121,14 @@ def to_queries(self) -> List[ast.SelectQuery | ast.SelectUnionQuery]:
                 else:
                     query_date_range = self.query_previous_date_range
 
-                query_builder: TrendsQueryBuilderAbstract
-
-                if isinstance(series.series, DataWarehouseNode):
-                    query_builder = DataWarehouseTrendsQueryBuilder(
-                        trends_query=series.overriden_query or self.query,
-                        team=self.team,
-                        query_date_range=query_date_range,
-                        series=series.series,
-                        timings=self.timings,
-                        modifiers=self.modifiers,
-                    )
-                else:
-                    query_builder = TrendsQueryBuilder(
-                        trends_query=series.overriden_query or self.query,
-                        team=self.team,
-                        query_date_range=query_date_range,
-                        series=series.series,
-                        timings=self.timings,
-                        modifiers=self.modifiers,
-                    )
+                query_builder = TrendsQueryBuilder(
+                    trends_query=series.overriden_query or self.query,
+                    team=self.team,
+                    query_date_range=query_date_range,
+                    series=series.series,
+                    timings=self.timings,
+                    modifiers=self.modifiers,
+                )
                 query = query_builder.build_query()
 
                 # Get around the default 100 limit, bump to the max 10000.
@@ -569,6 +556,20 @@ def update_hogql_modifiers(self) -> None:
         ):
             self.modifiers.inCohortVia = InCohortVia.leftjoin_conjoined
 
+        datawarehouse_modifiers = []
+        for series in self.query.series:
+            if isinstance(series, DataWarehouseNode):
+                datawarehouse_modifiers.append(
+                    DataWarehouseEventsModifier(
+                        table_name=series.table_name,
+                        timestamp_field=series.timestamp_field,
+                        id_field=series.id_field,
+                        distinct_id_field=series.distinct_id_field,
+                    )
+                )
+
+        self.modifiers.dataWarehouseEventsModifiers = datawarehouse_modifiers
+
     def setup_series(self) -> List[SeriesWithExtras]:
         series_with_extras = [
             SeriesWithExtras(
diff --git a/posthog/schema.py b/posthog/schema.py
index 5041c7f582ffe..dc77da163db17 100644
--- a/posthog/schema.py
+++ b/posthog/schema.py
@@ -151,6 +151,16 @@ class CountPerActorMathType(str, Enum):
     p99_count_per_actor = "p99_count_per_actor"
 
 
+class DataWarehouseEventsModifier(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    distinct_id_field: str
+    id_field: str
+    table_name: str
+    timestamp_field: str
+
+
 class DatabaseSchemaQueryResponseField(BaseModel):
     model_config = ConfigDict(
         extra="forbid",
@@ -414,6 +424,7 @@ class HogQLQueryModifiers(BaseModel):
     model_config = ConfigDict(
         extra="forbid",
     )
+    dataWarehouseEventsModifiers: Optional[List[DataWarehouseEventsModifier]] = None
     inCohortVia: Optional[InCohortVia] = None
     materializationMode: Optional[MaterializationMode] = None
     personsArgMaxVersion: Optional[PersonsArgMaxVersion] = None
@@ -1740,6 +1751,7 @@ class DataWarehouseNode(BaseModel):
         extra="forbid",
     )
     custom_name: Optional[str] = None
+    distinct_id_field: str
     fixedProperties: Optional[
         List[
             Union[