defmodule Plausible.Stats.Imported do @moduledoc """ Module defining functions that merge imported query into a native one, and also functions that decide whether imported data is eligible for a given query. """ use Plausible.ClickhouseRepo use Plausible.Stats.SQL.Fragments import Ecto.Query import Plausible.Stats.Imported.SQL.Expression alias Plausible.Stats.Imported alias Plausible.Stats.Query alias Plausible.Stats.SQL.QueryBuilder @property_to_table_mappings Imported.Base.property_to_table_mappings() @doc """ Returns a boolean indicating whether the combination of filters and breakdown property is possible to query from the imported tables. Usually, when no filters are used, the imported schema supports the query. There is one exception though - breakdown by a custom property. We are currently importing only two custom properties - `url` and `path`. Both these properties can only be used with their particular `event:name` filter or the corresponding goal filter (see Plausible.Event.SystemEvents). """ def schema_supports_query?(query) do length(Imported.Base.decide_tables(query)) > 0 end def schema_supports_interval?(query) do "time:minute" not in query.dimensions and "time:hour" not in query.dimensions end def merge_imported_country_suggestions(native_q, _site, %Plausible.Stats.Query{ include_imported: false }) do native_q end def merge_imported_country_suggestions(native_q, site, query) do supports_filter_set? = Enum.all?(query.filters, fn filter -> [_, filtered_prop | _] = filter @property_to_table_mappings[filtered_prop] == "imported_locations" end) if supports_filter_set? do native_q = native_q |> exclude(:order_by) |> exclude(:select) |> select([e], %{country_code: e.country_code, count: fragment("count(*)")}) imported_q = from i in Imported.Base.query_imported("imported_locations", site, query), group_by: i.country, select_merge: %{country_code: i.country, count: fragment("sum(?)", i.pageviews)} from(s in subquery(native_q), full_join: i in subquery(imported_q), on: s.country_code == i.country_code, select: fragment("if(not empty(?), ?, ?)", s.country_code, s.country_code, i.country_code), order_by: [desc: fragment("? + ?", s.count, i.count)] ) else native_q end end def merge_imported_region_suggestions(native_q, _site, %Plausible.Stats.Query{ include_imported: false }) do native_q end def merge_imported_region_suggestions(native_q, site, query) do supports_filter_set? = Enum.all?(query.filters, fn filter -> [_, filtered_prop | _] = filter @property_to_table_mappings[filtered_prop] == "imported_locations" end) if supports_filter_set? do native_q = native_q |> exclude(:order_by) |> exclude(:select) |> select([e], %{region_code: e.subdivision1_code, count: fragment("count(*)")}) imported_q = from i in Imported.Base.query_imported("imported_locations", site, query), where: i.region != "", group_by: i.region, select_merge: %{region_code: i.region, count: fragment("sum(?)", i.pageviews)} from(s in subquery(native_q), full_join: i in subquery(imported_q), on: s.region_code == i.region_code, select: fragment("if(not empty(?), ?, ?)", s.region_code, s.region_code, i.region_code), order_by: [desc: fragment("? + ?", s.count, i.count)] ) else native_q end end def merge_imported_city_suggestions(native_q, _site, %Plausible.Stats.Query{ include_imported: false }) do native_q end def merge_imported_city_suggestions(native_q, site, query) do supports_filter_set? = Enum.all?(query.filters, fn filter -> [_, filtered_prop | _] = filter @property_to_table_mappings[filtered_prop] == "imported_locations" end) if supports_filter_set? do native_q = native_q |> exclude(:order_by) |> exclude(:select) |> select([e], %{city_id: e.city_geoname_id, count: fragment("count(*)")}) imported_q = from i in Imported.Base.query_imported("imported_locations", site, query), where: i.city != 0, group_by: i.city, select_merge: %{city_id: i.city, count: fragment("sum(?)", i.pageviews)} from(s in subquery(native_q), full_join: i in subquery(imported_q), on: s.city_id == i.city_id, select: fragment("if(? > 0, ?, ?)", s.city_id, s.city_id, i.city_id), order_by: [desc: fragment("? + ?", s.count, i.count)] ) else native_q end end def merge_imported_filter_suggestions( native_q, _site, %Plausible.Stats.Query{include_imported: false}, _filter_name, _filter_search ) do native_q end def merge_imported_filter_suggestions( native_q, site, query, filter_name, filter_query ) do {table, db_field} = expand_suggestions_field(filter_name) supports_filter_set? = Enum.all?(query.filters, fn filter -> [_, filtered_prop | _] = filter @property_to_table_mappings[filtered_prop] == table end) if supports_filter_set? do native_q = native_q |> exclude(:order_by) |> exclude(:select) |> select([e], %{name: field(e, ^filter_name), count: fragment("count(*)")}) imported_q = from i in Imported.Base.query_imported(table, site, query), where: fragment("? ilike ?", field(i, ^db_field), ^filter_query), group_by: field(i, ^db_field), select_merge: %{name: field(i, ^db_field), count: fragment("sum(?)", i.pageviews)} from(s in subquery(native_q), full_join: i in subquery(imported_q), on: s.name == i.name, select: fragment("if(not empty(?), ?, ?)", s.name, s.name, i.name), order_by: [desc: fragment("? + ?", s.count, i.count)], limit: 25 ) else native_q end end @filter_suggestions_mapping %{ referrer_source: :source, acquisition_channel: :channel, screen_size: :device, pathname: :page } defp expand_suggestions_field(filter_name) do db_field = Map.get(@filter_suggestions_mapping, filter_name, filter_name) property = case db_field do :operating_system -> :os :operating_system_version -> :os_version other -> other end table_by_visit = Map.get(@property_to_table_mappings, "visit:#{property}") table_by_event = Map.get(@property_to_table_mappings, "event:#{property}") table = table_by_visit || table_by_event {table, db_field} end def merge_imported(q, _, %Query{include_imported: false}), do: q def merge_imported(q, site, %Query{dimensions: []} = query) do q = paginate_optimization(q, query) imported_q = site |> Imported.Base.query_imported(query) |> select_imported_metrics(query) |> paginate_optimization(query) from( s in subquery(q), cross_join: i in subquery(imported_q), select: %{} ) |> select_joined_metrics(query) end def merge_imported(q, site, %Query{dimensions: ["event:goal"]} = query) do goal_join_data = Plausible.Stats.Goals.goal_join_data(query) Imported.Base.decide_tables(query) |> Enum.map(fn "imported_custom_events" -> Imported.Base.query_imported("imported_custom_events", site, query) |> where([i], i.visitors > 0) |> select_merge_as([i], %{ dim0: fragment( "indexOf(?, ?)", type(^goal_join_data.event_names_imports, {:array, :string}), i.name ) }) |> select_imported_metrics(query) |> group_by([], selected_as(:dim0)) |> where([], selected_as(:dim0) != 0) "imported_pages" -> Imported.Base.query_imported("imported_pages", site, query) |> where([i], i.visitors > 0) |> where( [i], fragment( """ notEmpty( arrayFilter( goal_idx -> ?[goal_idx] = 'page' AND match(?, ?[goal_idx]), ? ) as indices ) """, type(^goal_join_data.types, {:array, :string}), i.page, type(^goal_join_data.page_regexes, {:array, :string}), type(^goal_join_data.indices, {:array, :integer}) ) ) |> join(:inner, [_i], index in fragment("indices"), hints: "ARRAY", on: true) |> group_by([_i, index], index) |> select_merge_as([_i, index], %{ dim0: fragment("CAST(?, 'UInt64')", index) }) |> select_imported_metrics(query) end) |> Enum.reduce(q, fn imports_q, q -> naive_dimension_join(q, imports_q, query) end) end def merge_imported(q, site, query) do if schema_supports_query?(query) do q = paginate_optimization(q, query) imported_q = site |> Imported.Base.query_imported(query) |> where([i], i.visitors > 0) |> group_imported_by(query) |> select_imported_metrics(query) |> paginate_optimization(query) from(s in subquery(q), full_join: i in subquery(imported_q), on: ^QueryBuilder.build_group_by_join(query), select: %{} ) |> select_joined_dimensions(query) |> select_joined_metrics(query) else q end end def total_imported_visitors(site, query) do site |> Imported.Base.query_imported(query) |> select_merge([i], %{total_visitors: fragment("sum(?)", i.visitors)}) end defp naive_dimension_join(q1, q2, query) do from(a in subquery(q1), full_join: b in subquery(q2), on: a.dim0 == b.dim0, select: %{} ) |> select_merge_as([a, b], %{ dim0: fragment("if(? != 0, ?, ?)", a.dim0, a.dim0, b.dim0) }) |> select_joined_metrics(query) end # Optimization for cases when grouping by a very high cardinality column. # # Instead of joining all rows from main and imported tables, we limit the number of rows # in both tables to LIMIT N * 100. # # This speeds up cases where a site has millions of unique pathnames, reducing the time spent # JOINing tables by an order of magnitude. # # Note that this optimization is lossy as the true top N values can arise from outside the top C # items of either subquery. In practice though, this will give plausible results. # # We only apply this optimization in cases where we can deterministically ORDER BY. This covers # opening Plausible dashboard but not more complicated use-cases. defp paginate_optimization(q, query) do if is_map(query.pagination) and can_order_by?(query) do n = (query.pagination.limit + query.pagination.offset) * 100 q |> QueryBuilder.build_order_by(query) |> limit(^n) else q end end @cannot_optimize_metrics [ :exit_rate, :scroll_depth, :percentage, :bounce_rate, :conversion_rate, :group_conversion_rate, :time_on_page, :total_revenue, :average_revenue ] defp can_order_by?(query) do Enum.all?(query.order_by, fn {metric, _direction} when metric in @cannot_optimize_metrics -> false {metric, _direction} when is_atom(metric) -> metric in query.metrics _ -> true end) end end