diff --git a/ee/api/dashboard_collaborator.py b/ee/api/dashboard_collaborator.py
index 0e070f0c2860c..998eeba8238f9 100644
--- a/ee/api/dashboard_collaborator.py
+++ b/ee/api/dashboard_collaborator.py
@@ -88,7 +88,7 @@ class DashboardCollaboratorViewSet(
     mixins.DestroyModelMixin,
     viewsets.GenericViewSet,
 ):
-    scope_object = "INTERNAL"
+    scope_object = "dashboard"
     permission_classes = [CanEditDashboardCollaborator]
     pagination_class = None
     queryset = DashboardPrivilege.objects.select_related("dashboard").filter(user__is_active=True)
diff --git a/frontend/src/lib/components/ActivityLog/__mocks__/activityLogMocks.ts b/frontend/src/lib/components/ActivityLog/__mocks__/activityLogMocks.ts
index 1727f0ed6bb8c..3026aa131089a 100644
--- a/frontend/src/lib/components/ActivityLog/__mocks__/activityLogMocks.ts
+++ b/frontend/src/lib/components/ActivityLog/__mocks__/activityLogMocks.ts
@@ -3,6 +3,36 @@ import { ActivityLogItem } from 'lib/components/ActivityLog/humanizeActivity'
 import { ActivityScope, InsightShortId } from '~/types'
 
 export const teamActivityResponseJson: ActivityLogItem[] = [
+    {
+        user: {
+            first_name: 'sdavasdadadsadas',
+            last_name: '',
+            email: 'paul@posthog.com',
+        },
+        unread: false,
+        is_staff: false,
+        is_system: false,
+        activity: 'updated',
+        item_id: '1',
+        scope: ActivityScope.TEAM,
+        detail: {
+            name: 'Default Project',
+            type: undefined,
+            merge: null,
+            changes: [
+                {
+                    type: ActivityScope.TEAM,
+                    after: { poe_v2_enabled: true },
+                    field: 'extra_settings',
+                    action: 'created',
+                    before: null,
+                },
+            ],
+            trigger: null,
+            short_id: null,
+        },
+        created_at: '2024-02-08T19:23:53.530402Z',
+    },
     {
         user: {
             first_name: 'sdavasdadadsadas',
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/scenes/onboarding/Onboarding.tsx b/frontend/src/scenes/onboarding/Onboarding.tsx
index e22ded649f5d3..569760e106bf6 100644
--- a/frontend/src/scenes/onboarding/Onboarding.tsx
+++ b/frontend/src/scenes/onboarding/Onboarding.tsx
@@ -100,6 +100,8 @@ const ProductAnalyticsOnboarding = (): JSX.Element => {
 }
 const SessionReplayOnboarding = (): JSX.Element => {
     const { hasAvailableFeature } = useValues(userLogic)
+    const { currentTeam } = useValues(teamLogic)
+
     const configOptions: ProductConfigOption[] = [
         {
             type: 'toggle',
@@ -107,7 +109,7 @@ const SessionReplayOnboarding = (): JSX.Element => {
             description: `Capture console logs as a part of user session recordings. 
                             Use the console logs alongside recordings to debug any issues with your app.`,
             teamProperty: 'capture_console_log_opt_in',
-            value: true,
+            value: currentTeam?.capture_console_log_opt_in ?? true,
         },
         {
             type: 'toggle',
@@ -115,7 +117,7 @@ const SessionReplayOnboarding = (): JSX.Element => {
             description: `Capture performance and network information alongside recordings. Use the
                             network requests and timings in the recording player to help you debug issues with your app.`,
             teamProperty: 'capture_performance_opt_in',
-            value: true,
+            value: currentTeam?.capture_performance_opt_in ?? true,
         },
     ]
 
@@ -126,7 +128,7 @@ const SessionReplayOnboarding = (): JSX.Element => {
             description: `Only record sessions that are longer than the specified duration. 
                             Start with it low and increase it later if you're getting too many short sessions.`,
             teamProperty: 'session_recording_minimum_duration_milliseconds',
-            value: null,
+            value: currentTeam?.session_recording_minimum_duration_milliseconds || null,
             selectOptions: SESSION_REPLAY_MINIMUM_DURATION_OPTIONS,
         })
     }
diff --git a/frontend/src/scenes/onboarding/OnboardingProductConfiguration.tsx b/frontend/src/scenes/onboarding/OnboardingProductConfiguration.tsx
index 5e7357d650450..8698b6bf64dbd 100644
--- a/frontend/src/scenes/onboarding/OnboardingProductConfiguration.tsx
+++ b/frontend/src/scenes/onboarding/OnboardingProductConfiguration.tsx
@@ -1,6 +1,6 @@
 import { LemonDivider, LemonSelect, LemonSwitch } from '@posthog/lemon-ui'
 import { useActions, useValues } from 'kea'
-import React, { useEffect } from 'react'
+import React, { useEffect, useRef } from 'react'
 import { pluginsLogic } from 'scenes/plugins/pluginsLogic'
 
 import { OnboardingStepKey } from './onboardingLogic'
@@ -51,6 +51,12 @@ export const OnboardingProductConfiguration = ({
     const { setConfigOptions, saveConfiguration } = useActions(onboardingProductConfigurationLogic)
     const { toggleEnabled } = useActions(pluginsLogic)
 
+    const configOptionsRef = useRef(configOptions)
+
+    useEffect(() => {
+        configOptionsRef.current = configOptions
+    }, [configOptions])
+
     useEffect(() => {
         setConfigOptions(options)
     }, [])
@@ -63,9 +69,12 @@ export const OnboardingProductConfiguration = ({
             selectOptions: option.selectOptions,
             value: option.value,
             onChange: (newValue: boolean | string | number) => {
-                setConfigOptions(
-                    configOptions.map((o) => (o.teamProperty === option.teamProperty ? { ...o, value: newValue } : o))
+                // Use the current value from the ref to ensure that onChange always accesses
+                // the latest state of configOptions, preventing the closure from using stale data.
+                const updatedConfigOptions = configOptionsRef.current.map((o) =>
+                    o.teamProperty === option.teamProperty ? { ...o, value: newValue } : o
                 )
+                setConfigOptions(updatedConfigOptions)
             },
         })),
         ...defaultEnabledPlugins.map((plugin) => {
diff --git a/frontend/src/scenes/teamActivityDescriber.tsx b/frontend/src/scenes/teamActivityDescriber.tsx
index a58fd574c3ebd..e9cb75390c4b0 100644
--- a/frontend/src/scenes/teamActivityDescriber.tsx
+++ b/frontend/src/scenes/teamActivityDescriber.tsx
@@ -168,7 +168,19 @@ const teamActionsMapping: Record<
     timezone: () => null,
     surveys_opt_in: () => null,
     week_start_day: () => null,
-    extra_settings: () => null,
+    extra_settings: (change: ActivityChange | undefined): ChangeMapping | null => {
+        const after = change?.after
+        if (typeof after !== 'object') {
+            return null
+        }
+        const descriptions = []
+        for (const key in after) {
+            if (key === 'poe_v2_enabled') {
+                descriptions.push(<>{after[key] ? 'enabled' : 'disabled'} Person on Events (v2)</>)
+            }
+        }
+        return { description: descriptions }
+    },
     has_completed_onboarding_for: () => null,
     // should never come from the backend
     created_at: () => null,
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/package.json b/package.json
index 49cee8126a85b..7c39d65568a7b 100644
--- a/package.json
+++ b/package.json
@@ -142,7 +142,7 @@
         "pmtiles": "^2.11.0",
         "postcss": "^8.4.31",
         "postcss-preset-env": "^9.3.0",
-        "posthog-js": "1.115.1",
+        "posthog-js": "1.115.2",
         "posthog-js-lite": "2.5.0",
         "prettier": "^2.8.8",
         "prop-types": "^15.7.2",
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 24e034e9b45bb..78423f993f9b0 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -245,8 +245,8 @@ dependencies:
     specifier: ^9.3.0
     version: 9.3.0(postcss@8.4.31)
   posthog-js:
-    specifier: 1.115.1
-    version: 1.115.1
+    specifier: 1.115.2
+    version: 1.115.2
   posthog-js-lite:
     specifier: 2.5.0
     version: 2.5.0
@@ -17252,8 +17252,8 @@ packages:
     resolution: {integrity: sha512-Urvlp0Vu9h3td0BVFWt0QXFJDoOZcaAD83XM9d91NKMKTVPZtfU0ysoxstIf5mw/ce9ZfuMgpWPaagrZI4rmSg==}
     dev: false
 
-  /posthog-js@1.115.1:
-    resolution: {integrity: sha512-920eIQlMJ/o7RcDAzRkAwb2+eMYhtzAJtL7zDwoDFLIgCGQ/313ClRywhVtX+o0FExypfqUWjH6rWCGsP9mCAQ==}
+  /posthog-js@1.115.2:
+    resolution: {integrity: sha512-nGTxDjH8df0FTd1plIqKFsmSynkkI/LmvYlJP7sqeKvtXhcQpVi4+avMhNWIasoWvyQbp65hmvwXyXyQ7jk2cw==}
     dependencies:
       fflate: 0.4.8
       preact: 10.19.6
diff --git a/posthog/api/organization_invite.py b/posthog/api/organization_invite.py
index 72345f6960759..cebf7e1f61e2d 100644
--- a/posthog/api/organization_invite.py
+++ b/posthog/api/organization_invite.py
@@ -83,7 +83,7 @@ class OrganizationInviteViewSet(
     mixins.ListModelMixin,
     viewsets.GenericViewSet,
 ):
-    scope_object = "INTERNAL"
+    scope_object = "organization_member"
     serializer_class = OrganizationInviteSerializer
     queryset = OrganizationInvite.objects.all()
     lookup_field = "id"
@@ -96,7 +96,7 @@ def get_queryset(self):
             .order_by(self.ordering)
         )
 
-    @action(methods=["POST"], detail=False)
+    @action(methods=["POST"], detail=False, required_scopes=["organization_member:write"])
     def bulk(self, request: request.Request, **kwargs) -> response.Response:
         data = cast(Any, request.data)
         if not isinstance(data, list):
diff --git a/posthog/api/test/__snapshots__/test_api_docs.ambr b/posthog/api/test/__snapshots__/test_api_docs.ambr
index c378bbb274781..728dabc13af0e 100644
--- a/posthog/api/test/__snapshots__/test_api_docs.ambr
+++ b/posthog/api/test/__snapshots__/test_api_docs.ambr
@@ -26,6 +26,7 @@
     '/home/runner/work/posthog/posthog/posthog/api/dashboards/dashboard_templates.py: Warning [DashboardTemplateViewSet]: could not derive type of path parameter "project_id" because it is untyped and obtaining queryset from the viewset failed. Consider adding a type to the path (e.g. <int:project_id>) or annotating the parameter type with @extend_schema. Defaulting to "string".',
     '/home/runner/work/posthog/posthog/posthog/api/dashboards/dashboard_templates.py: Warning [DashboardTemplateViewSet]: could not derive type of path parameter "id" because it is untyped and obtaining queryset from the viewset failed. Consider adding a type to the path (e.g. <int:id>) or annotating the parameter type with @extend_schema. Defaulting to "string".',
     '/home/runner/work/posthog/posthog/posthog/api/dashboards/dashboard.py: Warning [DashboardsViewSet]: could not derive type of path parameter "project_id" because model "posthog.models.dashboard.Dashboard" contained no such field. Consider annotating parameter with @extend_schema. Defaulting to "string".',
+    '/home/runner/work/posthog/posthog/ee/api/dashboard_collaborator.py: Warning [DashboardCollaboratorViewSet]: could not derive type of path parameter "project_id" because model "ee.models.dashboard_privilege.DashboardPrivilege" contained no such field. Consider annotating parameter with @extend_schema. Defaulting to "string".',
     '/home/runner/work/posthog/posthog/posthog/api/sharing.py: Warning [SharingConfigurationViewSet]: could not derive type of path parameter "project_id" because model "posthog.models.sharing_configuration.SharingConfiguration" contained no such field. Consider annotating parameter with @extend_schema. Defaulting to "string".',
     '/home/runner/work/posthog/posthog/posthog/api/early_access_feature.py: Warning [EarlyAccessFeatureViewSet]: could not derive type of path parameter "project_id" because model "posthog.models.early_access_feature.EarlyAccessFeature" contained no such field. Consider annotating parameter with @extend_schema. Defaulting to "string".',
     "/home/runner/work/posthog/posthog/posthog/api/event_definition.py: Error [EventDefinitionViewSet]: exception raised while getting serializer. Hint: Is get_serializer_class() returning None or is get_queryset() not working without a request? Ignoring the view for now. (Exception: 'AnonymousUser' object has no attribute 'organization')",
@@ -88,6 +89,8 @@
     '/home/runner/work/posthog/posthog/posthog/api/survey.py: Warning [SurveyViewSet]: could not derive type of path parameter "project_id" because model "posthog.models.feedback.survey.Survey" contained no such field. Consider annotating parameter with @extend_schema. Defaulting to "string".',
     'Warning: encountered multiple names for the same choice set (HrefMatchingEnum). This may be unwanted even though the generated schema is technically correct. Add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
     'Warning: enum naming encountered a non-optimally resolvable collision for fields named "type". The same name has been used for multiple choice sets in multiple components. The collision was resolved with "Type7baEnum". add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
+    'Warning: enum naming encountered a non-optimally resolvable collision for fields named "level". The same name has been used for multiple choice sets in multiple components. The collision was resolved with "LevelD7eEnum". add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
+    'Warning: encountered multiple names for the same choice set (RestrictionLevelEnum). This may be unwanted even though the generated schema is technically correct. Add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
     'Warning: encountered multiple names for the same choice set (EffectivePrivilegeLevelEnum). This may be unwanted even though the generated schema is technically correct. Add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
     'Warning: encountered multiple names for the same choice set (MembershipLevelEnum). This may be unwanted even though the generated schema is technically correct. Add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
     'Warning: operationId "list" has collisions [(\'/api/organizations/\', \'get\'), (\'/api/organizations/{organization_id}/projects/\', \'get\')]. resolving with numeral suffixes.',
diff --git a/posthog/api/test/test_team.py b/posthog/api/test/test_team.py
index d23efe81cf7d8..11bc5c664e1dd 100644
--- a/posthog/api/test/test_team.py
+++ b/posthog/api/test/test_team.py
@@ -204,6 +204,47 @@ def test_update_project_timezone(self):
             ]
         )
 
+    @freeze_time("2022-02-08")
+    def test_activity_log_tracks_extra_settings(self):
+        self._assert_activity_log_is_empty()
+
+        response = self.client.patch("/api/projects/@current/", {"extra_settings": {"poe_v2_enabled": True}})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        response_data = response.json()
+        self.assertEqual(response_data["name"], self.team.name)
+        self.assertEqual(response_data["extra_settings"], {"poe_v2_enabled": True})
+
+        self._assert_activity_log(
+            [
+                {
+                    "activity": "updated",
+                    "created_at": "2022-02-08T00:00:00Z",
+                    "detail": {
+                        "changes": [
+                            {
+                                "action": "created",
+                                "after": {"poe_v2_enabled": True},
+                                "before": None,
+                                "field": "extra_settings",
+                                "type": "Team",
+                            },
+                        ],
+                        "name": "Default project",
+                        "short_id": None,
+                        "trigger": None,
+                        "type": None,
+                    },
+                    "item_id": str(self.team.pk),
+                    "scope": "Team",
+                    "user": {
+                        "email": "user1@posthog.com",
+                        "first_name": "",
+                    },
+                },
+            ]
+        )
+
     def test_update_test_filter_default_checked(self):
         response = self.client.patch("/api/projects/@current/", {"test_account_filters_default_checked": "true"})
         self.assertEqual(response.status_code, status.HTTP_200_OK)
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/hogql_queries/legacy_compatibility/filter_to_query.py b/posthog/hogql_queries/legacy_compatibility/filter_to_query.py
index fc68d3f26c2f4..2b8f59f88a421 100644
--- a/posthog/hogql_queries/legacy_compatibility/filter_to_query.py
+++ b/posthog/hogql_queries/legacy_compatibility/filter_to_query.py
@@ -25,6 +25,7 @@
     StickinessQuery,
     TrendsFilter,
     TrendsQuery,
+    FunnelVizType,
 )
 from posthog.types import InsightQueryNode
 
@@ -392,9 +393,15 @@ def _insight_filter(filter: Dict):
             )
         }
     elif _insight_type(filter) == "FUNNELS":
+        funnel_viz_type = filter.get("funnel_viz_type")
+        # Backwards compatibility
+        # Before Filter.funnel_viz_type funnel trends were indicated by Filter.display being TRENDS_LINEAR
+        if funnel_viz_type is None and filter.get("display") == "ActionsLineGraph":
+            funnel_viz_type = FunnelVizType.trends
+
         insight_filter = {
             "funnelsFilter": FunnelsFilter(
-                funnelVizType=filter.get("funnel_viz_type"),
+                funnelVizType=funnel_viz_type,
                 funnelOrderType=filter.get("funnel_order_type"),
                 funnelFromStep=filter.get("funnel_from_step"),
                 funnelToStep=filter.get("funnel_to_step"),
@@ -416,12 +423,16 @@ def _insight_filter(filter: Dict):
                 retentionType=filter.get("retention_type"),
                 retentionReference=filter.get("retention_reference"),
                 totalIntervals=filter.get("total_intervals"),
-                returningEntity=to_base_entity_dict(filter.get("returning_entity"))
-                if filter.get("returning_entity") is not None
-                else None,
-                targetEntity=to_base_entity_dict(filter.get("target_entity"))
-                if filter.get("target_entity") is not None
-                else None,
+                returningEntity=(
+                    to_base_entity_dict(filter.get("returning_entity"))
+                    if filter.get("returning_entity") is not None
+                    else None
+                ),
+                targetEntity=(
+                    to_base_entity_dict(filter.get("target_entity"))
+                    if filter.get("target_entity") is not None
+                    else None
+                ),
                 period=filter.get("period"),
             )
         }
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[