WIP on dynamic refactor

Author: zachdaniel

lib/aggregate.ex

@@ -0,0 +1,136 @@
+defmodule AshPostgres.Sort do
+  @moduledoc false
+  require Ecto.Query
+
+  def sort(query, sort, resource) do
+    {:ok, query}
+    query = AshPostgres.DataLayer.default_bindings(query, resource)
+
+    sort
+    |> sanitize_sort()
+    |> Enum.reduce_while({:ok, []}, fn
+      {order, %Ash.Query.Calculation{} = calc}, {:ok, query_expr} ->
+        type =
+          if calc.type do
+            AshPostgres.Types.parameterized_type(calc.type, [])
+          else
+            nil
+          end
+
+        calc.opts
+        |> calc.module.expression(calc.context)
+        |> Ash.Filter.hydrate_refs(%{
+          resource: resource,
+          aggregates: query.__ash_bindings__.aggregate_defs,
+          calculations: %{},
+          public?: false
+        })
+        |> case do
+          {:ok, expr} ->
+            expr = AshPostgres.Expr.dynamic_expr(expr, query.__ash_bindings__, false, type)
+
+            {:cont, {:ok, query_expr ++ [{order, expr}]}}
+
+          {:error, error} ->
+            {:halt, {:error, error}}
+        end
+
+      {order, sort}, {:ok, query_expr} ->
+        expr =
+          case Map.fetch(query.__ash_bindings__.aggregates, sort) do
+            {:ok, binding} ->
+              aggregate =
+                Ash.Resource.Info.aggregate(resource, sort) ||
+                  raise "No such aggregate for query aggregate #{inspect(sort)}"
+
+              {:ok, field_type} =
+                if aggregate.field do
+                  related = Ash.Resource.Info.related(resource, aggregate.relationship_path)
+
+                  attr = Ash.Resource.Info.attribute(related, aggregate.field)
+
+                  if attr && related do
+                    {:ok, AshPostgres.Types.parameterized_type(attr.type, attr.constraints)}
+                  else
+                    {:ok, nil}
+                  end
+                else
+                  {:ok, nil}
+                end
+
+              default_value =
+                aggregate.default || Ash.Query.Aggregate.default_value(aggregate.kind)
+
+              if is_nil(default_value) do
+                Ecto.Query.dynamic(field(as(^binding), ^sort))
+              else
+                if field_type do
+                  Ecto.Query.dynamic(
+                    coalesce(field(as(^binding), ^sort), type(^default_value, ^field_type))
+                  )
+                else
+                  Ecto.Query.dynamic(coalesce(field(as(^binding), ^sort), ^default_value))
+                end
+              end
+
+            :error ->
+              Ecto.Query.dynamic(field(as(^0), ^sort))
+          end
+
+        {:cont, {:ok, query_expr ++ [{order, expr}]}}
+    end)
+    |> case do
+      {:ok, []} ->
+        {:ok, query}
+
+      {:ok, sort_exprs} ->
+        new_query = Ecto.Query.order_by(query, ^sort_exprs)
+
+        sort_expr = List.last(new_query.order_bys)
+
+        new_query =
+          new_query
+          |> Map.update!(:windows, fn windows ->
+            order_by_expr = %{sort_expr | expr: [order_by: sort_expr.expr]}
+            Keyword.put(windows, :order, order_by_expr)
+          end)
+
+        {:ok, new_query}
+
+      {:error, error} ->
+        {:error, error}
+    end
+  end
+
+  def order_to_postgres_order(dir) do
+    case dir do
+      :asc -> nil
+      :asc_nils_last -> " ASC NULLS LAST"
+      :asc_nils_first -> " ASC NULLS FIRST"
+      :desc -> " DESC"
+      :desc_nils_last -> " DESC NULLS LAST"
+      :desc_nils_first -> " DESC NULLS FIRST"
+    end
+  end
+
+  defp sanitize_sort(sort) do
+    sort
+    |> List.wrap()
+    |> Enum.map(fn
+      {sort, {order, context}} ->
+        {ash_to_ecto_order(order), {sort, context}}
+
+      {sort, order} ->
+        {ash_to_ecto_order(order), sort}
+
+      sort ->
+        sort
+    end)
+  end
+
+  defp ash_to_ecto_order(:asc_nils_last), do: :asc_nulls_last
+  defp ash_to_ecto_order(:asc_nils_first), do: :asc_nulls_first
+  defp ash_to_ecto_order(:desc_nils_last), do: :desc_nulls_last
+  defp ash_to_ecto_order(:desc_nils_first), do: :desc_nulls_first
+  defp ash_to_ecto_order(other), do: other
+end

lib/data_layer.ex

@@ -0,0 +1,89 @@
+defmodule AshPostgres.Calculation do
+  @moduledoc false
+
+  import Ecto.Query, only: [from: 2]
+
+  def add_calculation(query, calculation, expression, resource) do
+    query = AshPostgres.DataLayer.default_bindings(query, resource)
+
+    query =
+      if query.select do
+        query
+      else
+        from(row in query,
+          select: row,
+          select_merge: %{aggregates: %{}, calculations: %{}}
+        )
+      end
+
+    expr =
+      AshPostgres.Expr.dynamic_expr(
+        expression,
+        query.__ash_bindings__
+      )
+
+    {:ok,
+     query
+     |> Map.update!(:select, &add_to_calculation_select(&1, expr, calculation))}
+  end
+
+  defp add_to_calculation_select(
+         %{
+           expr:
+             {:merge, _,
+              [
+                first,
+                {:%{}, _,
+                 [{:aggregates, {:%{}, [], agg_fields}}, {:calculations, {:%{}, [], fields}}]}
+              ]}
+         } = select,
+         expr,
+         %{load: nil} = calculation
+       ) do
+    field = expr |> IO.inspect()
+
+    name =
+      if calculation.sequence == 0 do
+        calculation.name
+      else
+        String.to_existing_atom("#{calculation.name}_#{calculation.sequence}")
+      end
+
+    new_fields = [
+      {name, field}
+      | fields
+    ]
+
+    %{
+      select
+      | expr:
+          {:merge, [],
+           [
+             first,
+             {:%{}, [],
+              [{:aggregates, {:%{}, [], agg_fields}}, {:calculations, {:%{}, [], new_fields}}]}
+           ]}
+    }
+  end
+
+  defp add_to_calculation_select(
+         %{expr: select_expr} = select,
+         expr,
+         %{load: load_as} = calculation
+       ) do
+    field =
+      Ecto.Query.dynamic(type(^expr, ^AshPostgres.Types.parameterized_type(calculation.type, [])))
+
+    load_as =
+      if calculation.sequence == 0 do
+        load_as
+      else
+        "#{load_as}_#{calculation.sequence}"
+      end
+
+    %{
+      select
+      | expr: {:merge, [], [select_expr, {:%{}, [], [{load_as, field}]}]}
+    }
+  end
+end

lib/expr.ex

@@ -0,0 +1,146 @@
+defmodule AshPostgres.Types do
+  @moduledoc false
+
+  alias Ash.Query.Ref
+
+  def parameterized_type({:array, type}, constraints) do
+    {:array, parameterized_type(type, constraints[:items] || [])}
+  end
+
+  def parameterized_type(Ash.Type.CiString, constraints) do
+    parameterized_type(Ash.Type.CiStringWrapper, constraints)
+  end
+
+  def parameterized_type(type, constraints) do
+    if Ash.Type.ash_type?(type) do
+      parameterized_type(Ash.Type.ecto_type(type), constraints)
+    else
+      if is_atom(type) && :erlang.function_exported(type, :type, 1) do
+        {:parameterized, type, constraints}
+      else
+        type
+      end
+    end
+  end
+
+  def determine_types(mod, values) do
+    Code.ensure_compiled(mod)
+
+    cond do
+      :erlang.function_exported(mod, :types, 0) ->
+        mod.types()
+
+      :erlang.function_exported(mod, :args, 0) ->
+        mod.args()
+
+      true ->
+        [:any]
+    end
+    |> Enum.map(fn types ->
+      case types do
+        :same ->
+          types =
+            for _ <- values do
+              :same
+            end
+
+          closest_fitting_type(types, values)
+
+        :any ->
+          for _ <- values do
+            :any
+          end
+
+        types ->
+          closest_fitting_type(types, values)
+      end
+    end)
+    |> Enum.min_by(fn types ->
+      types
+      |> Enum.map(&vagueness/1)
+      |> Enum.sum()
+    end)
+  end
+
+  defp closest_fitting_type(types, values) do
+    types_with_values = Enum.zip(types, values)
+
+    types_with_values
+    |> fill_in_known_types()
+    |> clarify_types()
+  end
+
+  defp clarify_types(types) do
+    basis =
+      types
+      |> Enum.map(&elem(&1, 0))
+      |> Enum.min_by(&vagueness(&1))
+
+    Enum.map(types, fn {type, _value} ->
+      replace_same(type, basis)
+    end)
+  end
+
+  defp replace_same({:in, type}, basis) do
+    {:in, replace_same(type, basis)}
+  end
+
+  defp replace_same(:same, :same) do
+    :any
+  end
+
+  defp replace_same(:same, {:in, :same}) do
+    {:in, :any}
+  end
+
+  defp replace_same(:same, basis) do
+    basis
+  end
+
+  defp replace_same(other, _basis) do
+    other
+  end
+
+  defp fill_in_known_types(types) do
+    Enum.map(types, &fill_in_known_type/1)
+  end
+
+  defp fill_in_known_type(
+         {vague_type, %Ref{attribute: %{type: type, constraints: constraints}}} = ref
+       )
+       when vague_type in [:any, :same] do
+    if Ash.Type.ash_type?(type) do
+      type = type |> Ash.Type.ecto_type() |> parameterized_type(constraints) |> array_to_in()
+      {type, ref}
+    else
+      type =
+        if is_atom(type) && :erlang.function_exported(type, :type, 1) do
+          {:parameterized, type, []} |> array_to_in()
+        else
+          type |> array_to_in()
+        end
+
+      {type, ref}
+    end
+  end
+
+  defp fill_in_known_type(
+         {{:array, type}, %Ref{attribute: %{type: {:array, type}} = attribute} = ref}
+       ) do
+    {:in, fill_in_known_type({type, %{ref | attribute: %{attribute | type: type}}})}
+  end
+
+  defp fill_in_known_type({type, value}), do: {array_to_in(type), value}
+
+  defp array_to_in({:array, v}), do: {:in, array_to_in(v)}
+
+  defp array_to_in({:parameterized, type, constraints}),
+    do: {:parameterized, array_to_in(type), constraints}
+
+  defp array_to_in(v), do: v
+
+  defp vagueness({:in, type}), do: vagueness(type)
+  defp vagueness(:same), do: 2
+  defp vagueness(:any), do: 1
+  defp vagueness(_), do: 0
+end

lib/join.ex

@@ -93,7 +93,9 @@ defmodule AshPostgres.MixProject do
   # Run "mix help deps" to learn about dependencies.
   defp deps do
     [
-      {:ecto_sql, "~> 3.7"},
+      {:ecto_sql, "~> 3.7", override: true},
+      # {:ecto, github: "zachdaniel/ecto", branch: "dynamic-bindings", override: true},
+      {:ecto, path: "../ecto", override: true},
       {:jason, "~> 1.0"},
       {:postgrex, ">= 0.0.0"},
       {:ash, ash_version("~> 1.50 and >= 1.50.5")},