Fill in person filtering and reintegrate tests

ee/clickhouse/models/property.py

@@ -4,6 +4,7 @@
 from ee.clickhouse.models.cohort import format_cohort_table_name
 from ee.clickhouse.sql.cohort import COHORT_DISTINCT_ID_FILTER_SQL
 from ee.clickhouse.sql.events import EVENT_PROP_CLAUSE, SELECT_PROP_VALUES_SQL, SELECT_PROP_VALUES_SQL_WITH_FILTER
+from ee.clickhouse.sql.person import GET_DISTINCT_IDS_BY_PROPERTY_SQL
 from posthog.models.cohort import Cohort
 from posthog.models.property import Property
 from posthog.models.team import Team
@@ -20,6 +21,17 @@ def parse_prop_clauses(key: str, filters: List[Property], team: Team, prepend: s
             clause = COHORT_DISTINCT_ID_FILTER_SQL.format(table_name=format_cohort_table_name(cohort))
             final += "{cond} ({clause}) ".format(cond="AND distinct_id IN", clause=clause)
 
+        elif prop.type == "person":
+            prepend = "person"
+            filter = "(ep.key = %(k{prepend}_{idx})s) AND (ep.value {operator} %(v{prepend}_{idx})s)".format(
+                idx=idx, operator=get_operator(prop.operator), prepend=prepend
+            )
+            clause = GET_DISTINCT_IDS_BY_PROPERTY_SQL.format(filters=filter)
+            final += "{cond} ({clause}) ".format(cond="AND distinct_id IN", clause=clause)
+            params.update(
+                {"k{}_{}".format(prepend, idx): prop.key, "v{}_{}".format(prepend, idx): _pad_value(prop.value)}
+            )
+
         else:
             filter = "(ep.key = %(k{prepend}_{idx})s) AND (ep.value {operator} %(v{prepend}_{idx})s)".format(
                 idx=idx, operator=get_operator(prop.operator), prepend=prepend
@@ -29,10 +41,15 @@ def parse_prop_clauses(key: str, filters: List[Property], team: Team, prepend: s
             params.update(
                 {"k{}_{}".format(prepend, idx): prop.key, "v{}_{}".format(prepend, idx): _pad_value(prop.value)}
             )
+
     return final, params
 
 
 def _pad_value(val: str):
+
+    if val == "true" or val == "false" or val.isdigit():
+        return val
+
     if not val.startswith('"'):
         val = '"' + val
 

ee/clickhouse/queries/clickhouse_trends.py

@@ -57,6 +57,36 @@
 )
 """
 
+TOP_PERSON_PROPS_ARRAY_OF_KEY_SQL = """
+SELECT groupArray(value) FROM (
+    SELECT value, count(*) as count
+    FROM
+    events e 
+    INNER JOIN person_distinct_id pid ON e.distinct_id = pid.distinct_id
+    INNER JOIN
+        (
+            SELECT * FROM (
+                SELECT
+                id,
+                array_property_keys as key,
+                array_property_values as value
+                from (
+                    SELECT
+                        id,
+                        arrayMap(k -> toString(k.1), JSONExtractKeysAndValuesRaw(properties)) AS array_property_keys,
+                        arrayMap(k -> toString(k.2), JSONExtractKeysAndValuesRaw(properties)) AS array_property_values
+                    FROM person WHERE team_id = %(team_id)s
+                )
+                ARRAY JOIN array_property_keys, array_property_values
+            ) ep
+            WHERE key = %(key)s
+        ) ep ON person_id = ep.id WHERE e.team_id = %(team_id)s {parsed_date_from} {parsed_date_to}
+    GROUP BY value
+    ORDER BY count DESC
+    LIMIT %(limit)s
+)
+"""
+
 BREAKDOWN_QUERY_SQL = """
 SELECT groupArray(day_start), groupArray(count), breakdown_value FROM (
     SELECT SUM(total) as count, day_start, breakdown_value FROM (
@@ -104,13 +134,37 @@
 where team_id = %(team_id)s {event_filter} {filters} {parsed_date_from} {parsed_date_to} {actions_query}
 """
 
+BREAKDOWN_PERSON_PROP_JOIN_SQL = """
+INNER JOIN person_distinct_id pid ON e.distinct_id = pid.distinct_id
+INNER JOIN (
+    SELECT * FROM (
+        SELECT
+        id,
+        array_property_keys as key,
+        array_property_values as value
+        from (
+            SELECT
+                id,
+                arrayMap(k -> toString(k.1), JSONExtractKeysAndValuesRaw(properties)) AS array_property_keys,
+                arrayMap(k -> toString(k.2), JSONExtractKeysAndValuesRaw(properties)) AS array_property_values
+            FROM person WHERE team_id = %(team_id)s
+        )
+        ARRAY JOIN array_property_keys, array_property_values
+    ) ep
+    WHERE key = %(key)s
+) ep 
+ON person_id = ep.id WHERE e.team_id = %(team_id)s {event_filter} {parsed_date_from} {parsed_date_to}
+AND breakdown_value in (%(values)s) {actions_query}
+"""
+
+
 BREAKDOWN_PROP_JOIN_SQL = """
 INNER JOIN (
     SELECT *
     FROM events_properties_view AS ep
     WHERE key = %(key)s and team_id = %(team_id)s
 ) ep 
-ON e.uuid = ep.event_id where team_id = %(team_id)s {event_filter} {filters} {parsed_date_from} {parsed_date_to}
+ON uuid = ep.event_id where e.team_id = %(team_id)s {event_filter} {parsed_date_from} {parsed_date_to}
 AND breakdown_value in (%(values)s) {actions_query}
 """
 
@@ -243,7 +297,35 @@ def _format_breakdown_query(self, entity: Entity, filter: Filter, team: Team) ->
                     aggregate_operation=aggregate_operation,
                 )
         elif filter.breakdown_type == "person":
-            pass
+            element_params = {**params, "key": filter.breakdown, "limit": 20}
+            element_query = TOP_PERSON_PROPS_ARRAY_OF_KEY_SQL.format(
+                parsed_date_from=parsed_date_from, parsed_date_to=parsed_date_to
+            )
+            try:
+                top_elements_array_result = sync_execute(element_query, element_params)
+                top_elements_array = top_elements_array_result[0][0]
+            except:
+                top_elements_array = []
+
+            params = {
+                **params,
+                "values": top_elements_array,
+                "key": filter.breakdown,
+                "event": entity.id,
+                **action_params,
+            }
+            breakdown_filter = BREAKDOWN_PERSON_PROP_JOIN_SQL.format(
+                parsed_date_from=parsed_date_from,
+                parsed_date_to=parsed_date_to,
+                actions_query="AND uuid IN ({})".format(action_query) if action_query else "",
+                event_filter="AND event = %(event)s" if not action_query else "",
+            )
+            breakdown_query = BREAKDOWN_QUERY_SQL.format(
+                null_sql=null_sql,
+                breakdown_filter=breakdown_filter,
+                event_join=join_condition,
+                aggregate_operation=aggregate_operation,
+            )
         else:
             element_params = {**params, "key": filter.breakdown, "limit": 20}
             element_query = TOP_ELEMENTS_ARRAY_OF_KEY_SQL.format(
@@ -285,17 +367,19 @@ def _format_breakdown_query(self, entity: Entity, filter: Filter, team: Team) ->
         parsed_results = []
 
         for idx, stats in enumerate(result):
-            extra_label = self._determine_breakdown_label(
-                idx, filter.breakdown_type, filter.breakdown, top_elements_array
-            )
+
+            breakdown_value = stats[2] if not filter.breakdown_type == "cohort" else ""
+            stripped_value = breakdown_value.strip('"') if isinstance(breakdown_value, str) else breakdown_value
+
+            extra_label = self._determine_breakdown_label(idx, filter.breakdown_type, filter.breakdown, stripped_value)
             label = "{} - {}".format(entity.name, extra_label)
             additional_values = {
                 "label": label,
                 "breakdown_value": filter.breakdown[idx]
                 if isinstance(filter.breakdown, list)
                 else filter.breakdown
                 if filter.breakdown_type == "cohort"
-                else top_elements_array[idx],
+                else stripped_value,
             }
             parsed_result = self._parse_response(stats, filter, additional_values)
             parsed_results.append(parsed_result)
@@ -350,18 +434,16 @@ def _determine_breakdown_label(
         index: int,
         breakdown_type: Optional[str],
         breakdown: Union[str, List[Union[str, int]], None],
-        elements: List,
+        value: Union[str, int],
     ) -> str:
         breakdown = breakdown if breakdown and isinstance(breakdown, list) else []
         if breakdown_type == "cohort":
             if breakdown[index] == "all":
                 return "all users"
             else:
                 return Cohort.objects.get(pk=breakdown[index]).name
-        elif breakdown_type == "person":
-            return ""
         else:
-            return str(elements[index]) or ""
+            return str(value) or ""
 
     def _process_math(self, entity):
         join_condition = ""

ee/clickhouse/queries/test/test_retention.py

@@ -14,8 +14,6 @@ def _create_event(**kwargs):
 
 
 class TestClickhouseRetention(ClickhouseTestMixin, retention_test_factory(ClickhouseRetention, _create_event, Person.objects.create)):  # type: ignore
-    def test_retention_with_user_properties(self):
-        pass
 
     # period filtering for clickhouse only
     def test_retention_period(self):

ee/clickhouse/queries/test/test_trends.py

@@ -39,6 +39,47 @@ def _create_event(**kwargs):
 
 
 class TestClickhouseTrends(ClickhouseTestMixin, trend_test_factory(ClickhouseTrends, _create_event, Person.objects.create, _create_action, _create_cohort)):  # type: ignore
+    def test_breakdown_by_person_property(self):
+        person1, person2, person3, person4 = self._create_multiple_people()
+        action = _create_action(name="watched movie", team=self.team)
+
+        with freeze_time("2020-01-04T13:01:01Z"):
+            action_response = ClickhouseTrends().run(
+                Filter(
+                    data={
+                        "date_from": "-14d",
+                        "breakdown": "name",
+                        "breakdown_type": "person",
+                        "actions": [{"id": action.pk, "type": "actions", "order": 0}],
+                    }
+                ),
+                self.team,
+            )
+            event_response = ClickhouseTrends().run(
+                Filter(
+                    data={
+                        "date_from": "-14d",
+                        "breakdown": "name",
+                        "breakdown_type": "person",
+                        "events": [{"id": "watched movie", "name": "watched movie", "type": "events", "order": 0,}],
+                    }
+                ),
+                self.team,
+            )
+
+        self.assertListEqual([res["breakdown_value"] for res in event_response], ["person1", "person2", "person3"])
+
+        for response in event_response:
+            if response["breakdown_value"] == "person1":
+                self.assertEqual(response["count"], 1)
+                self.assertEqual(response["label"], "watched movie - person1")
+            if response["breakdown_value"] == "person2":
+                self.assertEqual(response["count"], 3)
+            if response["breakdown_value"] == "person3":
+                self.assertEqual(response["count"], 3)
+
+        self.assertTrue(self._compare_entity_response(event_response, action_response,))
+
     def test_breakdown_filtering(self):
         self._create_events()
         # test breakdown filtering
@@ -57,13 +98,38 @@ def test_breakdown_filtering(self):
                 self.team,
             )
 
-        self.assertEqual(response[0]["label"], 'sign up - "value"')
-        self.assertEqual(response[1]["label"], 'sign up - "other_value"')
-        self.assertEqual(response[2]["label"], 'no events - "value"')
-        self.assertEqual(response[3]["label"], 'no events - "other_value"')
+        self.assertEqual(response[0]["label"], "sign up - value")
+        self.assertEqual(response[1]["label"], "sign up - other_value")
+        self.assertEqual(response[2]["label"], "no events - value")
+        self.assertEqual(response[3]["label"], "no events - other_value")
 
         self.assertEqual(sum(response[0]["data"]), 2)
-        self.assertEqual(response[0]["breakdown_value"], '"value"')
+        self.assertEqual(response[0]["breakdown_value"], "value")
 
         self.assertEqual(sum(response[1]["data"]), 1)
-        self.assertEqual(response[1]["breakdown_value"], '"other_value"')
+        self.assertEqual(response[1]["breakdown_value"], "other_value")
+
+    def test_dau_with_breakdown_filtering(self):
+        sign_up_action, _ = self._create_events()
+        with freeze_time("2020-01-02T13:01:01Z"):
+            _create_event(
+                team=self.team, event="sign up", distinct_id="blabla", properties={"$some_property": "other_value"},
+            )
+        with freeze_time("2020-01-04T13:01:01Z"):
+            action_response = ClickhouseTrends().run(
+                Filter(data={"breakdown": "$some_property", "actions": [{"id": sign_up_action.id, "math": "dau"}]}),
+                self.team,
+            )
+            event_response = ClickhouseTrends().run(
+                Filter(data={"breakdown": "$some_property", "events": [{"id": "sign up", "math": "dau"}]}), self.team,
+            )
+
+        self.assertEqual(event_response[0]["label"], "sign up - value")
+        self.assertEqual(event_response[1]["label"], "sign up - other_value")
+
+        self.assertEqual(sum(event_response[0]["data"]), 1)
+        self.assertEqual(event_response[0]["data"][3], 1)  # property not defined
+
+        self.assertEqual(sum(event_response[1]["data"]), 1)
+        self.assertEqual(event_response[1]["data"][4], 1)
+        self.assertTrue(self._compare_entity_response(action_response, event_response))

ee/clickhouse/sql/person.py

@@ -272,3 +272,25 @@
     ARRAY JOIN array_property_keys, array_property_values
 ) GROUP BY key ORDER BY count DESC LIMIT %(limit)s
 """
+
+
+GET_DISTINCT_IDS_BY_PROPERTY_SQL = """
+SELECT distinct_id FROM person_distinct_id where person_id IN 
+(
+    SELECT id FROM (
+        SELECT
+        id, 
+        array_property_keys as key,
+        array_property_values as value
+        from (
+            SELECT
+                id,
+                arrayMap(k -> toString(k.1), JSONExtractKeysAndValuesRaw(properties)) AS array_property_keys,
+                arrayMap(k -> toString(k.2), JSONExtractKeysAndValuesRaw(properties)) AS array_property_values
+            FROM person WHERE team_id = %(team_id)s
+        )
+        ARRAY JOIN array_property_keys, array_property_values
+    ) ep
+    WHERE {filters}
+)
+"""

posthog/queries/test/test_funnel.py

@@ -202,7 +202,7 @@ def test_funnel_prop_filters_per_entity(self):
                         "id": action_credit_card.pk,
                         "type": "actions",
                         "order": 1,
-                        "properties": [{"key": "$browser", "value": "Safari", "type": "person"}],
+                        "properties": [{"key": "$browser", "value": "Safari"}],
                     },
                     {
                         "id": action_play_movie.pk,