Repository: elixir-ecto/ecto
Pull request #2498 (2018-04-07 09:16:40)

Port select-only type validation logic to builder
solves #2496

@josevalim I was not sure how to write tests well, so I created scenario for my case as described in issue, so I'm casting type of json value in `order_by` macro for both adapters.

Hope I did not missed anything.

josevalim: @Eiji7 i have dropped some comments and this should be good to go!
Eiji7: @josevalim Is it ok now?

Patch:
diff --git a/lib/ecto/adapters/mysql/connection.ex b/lib/ecto/adapters/mysql/connection.ex
index b778788c..725b2d20 100644
--- a/lib/ecto/adapters/mysql/connection.ex
+++ b/lib/ecto/adapters/mysql/connection.ex
@@ -90,11 +90,15 @@ if Code.ensure_loaded?(Mariaex) do
       [select, from, join, where, group_by, having, order_by, limit, offset | lock]
     end
 
-    def update_all(query, prefix \\ nil)
+    def update_all(query, prefix \\ nil) do
+      %{from: %{source: source}, select: select} = query
+
+      if select do
+        error!(nil, ":select is not supported in update_all by MySQL")
+      end
 
-    def update_all(%{from: from, select: nil} = query, prefix) do
       sources = create_names(query)
-      {from, name} = get_source(query, sources, 0, from)
+      {from, name} = get_source(query, sources, 0, source)
 
       fields = if prefix do
         update_fields(:on_conflict, query, sources)
@@ -108,11 +112,12 @@ if Code.ensure_loaded?(Mariaex) do
 
       [prefix, fields | where]
     end
-    def update_all(_query, _prefix) do
-      error!(nil, ":select is not supported in update_all by MySQL")
-    end
 
     def delete_all(%{select: nil} = query) do
+      if query.select do
+        error!(nil, ":select is not supported in delete_all by MySQL")
+      end
+
       sources = create_names(query)
       {_, name, _} = elem(sources, 0)
 
@@ -122,8 +127,6 @@ if Code.ensure_loaded?(Mariaex) do
 
       ["DELETE ", name, ".*", from, join | where]
     end
-    def delete_all(_query),
-      do: error!(nil, ":select is not supported in delete_all by MySQL")
 
     def insert(prefix, table, header, rows, on_conflict, []) do
       fields = intersperse_map(header, ?,, &quote_name/1)
@@ -228,8 +231,8 @@ if Code.ensure_loaded?(Mariaex) do
       end)
     end
 
-    defp from(%{from: from} = query, sources) do
-      {from, name} = get_source(query, sources, 0, from)
+    defp from(%{from: %{source: source}} = query, sources) do
+      {from, name} = get_source(query, sources, 0, source)
       [" FROM ", from, " AS " | name]
     end
 
diff --git a/lib/ecto/adapters/postgres/connection.ex b/lib/ecto/adapters/postgres/connection.ex
index 7b889af6..99df53da 100644
--- a/lib/ecto/adapters/postgres/connection.ex
+++ b/lib/ecto/adapters/postgres/connection.ex
@@ -129,9 +129,9 @@ if Code.ensure_loaded?(Postgrex) do
       [select, from, join, where, group_by, having, order_by, limit, offset | lock]
     end
 
-    def update_all(%{from: from} = query, prefix \\ nil) do
+    def update_all(%{from: %{source: source}} = query, prefix \\ nil) do
       sources = create_names(query)
-      {from, name} = get_source(query, sources, 0, from)
+      {from, name} = get_source(query, sources, 0, source)
 
       prefix = prefix || ["UPDATE ", from, " AS ", name | " SET "]
       fields = update_fields(query, sources)
@@ -281,8 +281,8 @@ if Code.ensure_loaded?(Postgrex) do
        exprs}
     end
 
-    defp from(%{from: from} = query, sources) do
-      {from, name} = get_source(query, sources, 0, from)
+    defp from(%{from: %{source: source}} = query, sources) do
+      {from, name} = get_source(query, sources, 0, source)
       [" FROM ", from, " AS " | name]
     end
 
diff --git a/lib/ecto/association.ex b/lib/ecto/association.ex
index 7c87d765..25f4bd11 100644
--- a/lib/ecto/association.ex
+++ b/lib/ecto/association.ex
@@ -39,7 +39,7 @@ defmodule Ecto.Association do
                field: atom,
                unique: boolean}
 
-  alias Ecto.Query.{BooleanExpr, QueryExpr}
+  alias Ecto.Query.{BooleanExpr, QueryExpr, FromExpr}
 
   @doc """
   Builds the association struct.
@@ -175,7 +175,10 @@ defmodule Ecto.Association do
   end
 
   def assoc_query(refl, t, query, values) do
-    query = query || %Ecto.Query{from: {"join expression", nil}, prefix: refl.queryable.__schema__(:prefix)}
+    query = query || %Ecto.Query{
+      from: %FromExpr{source: {"join expression", nil}},
+      prefix: refl.queryable.__schema__(:prefix)
+    }
 
     # Find the position for upcoming joins
     position = length(query.joins) + 1
@@ -187,7 +190,8 @@ defmodule Ecto.Association do
     # Note we are being restrictive on the format
     # expected from assoc_query.
     assoc_query = refl.__struct__.assoc_query(refl, nil, values)
-    joins = Ecto.Query.Planner.query_to_joins(:inner, assoc_query, position)
+    %{from: %{source: assoc_source}} = assoc_query
+    joins = Ecto.Query.Planner.query_to_joins(:inner, assoc_source, assoc_query, position)
 
     # Add the new join to the query and traverse the remaining
     # joins that will start counting from the added join position.
@@ -229,8 +233,10 @@ defmodule Ecto.Association do
     |> Map.put(:op, :and)
   end
 
-  defp merge_from({"join expression", _}, assoc_source), do: assoc_source
-  defp merge_from(from, _assoc_source), do: from
+  defp merge_from(%FromExpr{source: {"join expression", _}} = from, assoc_source),
+    do: %{from | source: assoc_source}
+  defp merge_from(from, _assoc_source),
+    do: from
 
   # Rewrite all later joins
   defp rewrite_join(%{on: on, ix: ix} = join, mapping) when ix >= mapping do
@@ -273,15 +279,15 @@ defmodule Ecto.Association do
   """
   def related_from_query(atom, _name) when is_atom(atom), do: atom
   def related_from_query({source, schema}, _name) when is_binary(source) and is_atom(schema), do: schema
-  def related_from_query(%Ecto.Query{from: {source, schema}} = query, name) when is_binary(source) and is_atom(schema) do
+  def related_from_query(%Ecto.Query{from: %FromExpr{source: {_, schema}}} = query, name) when is_atom(schema) do
     case query do
       %Ecto.Query{order_bys: [], limit: nil, offset: nil, group_bys: [], joins: [],
                   havings: [], preloads: [], assocs: [], distinct: nil, lock: nil} ->
         schema
       _ ->
         raise ArgumentError,
-          "A query was provided for many_to_many #{inspect name}, but that query included a statement other" <>
-          "than a `where` clause. Queries in :join_through only support `where` clauses, nothing else."
+          "A query was provided for association #{inspect name}, but that query included a statement other" <>
+          "than a `where` clause. Queries in associations only support `where` clauses, nothing else."
     end
   end
   def related_from_query(queryable, name) do
diff --git a/lib/ecto/query.ex b/lib/ecto/query.ex
index 6e363adc..e649c96e 100644
--- a/lib/ecto/query.ex
+++ b/lib/ecto/query.ex
@@ -310,6 +310,11 @@ defmodule Ecto.Query do
              havings: [], preloads: [], assocs: [], distinct: nil, lock: nil]
   @type t :: %__MODULE__{}
 
+  defmodule FromExpr do
+    @moduledoc false
+    defstruct [:source]
+  end
+
   defmodule DynamicExpr do
     @moduledoc false
     defstruct [:fun, :binding, :file, :line]
@@ -1442,7 +1447,7 @@ defmodule Ecto.Query do
     %QueryExpr{expr: expr, file: __ENV__.file, line: __ENV__.line}
   end
 
-  defp assert_schema!(%{from: {_source, schema}}) when schema != nil, do: schema
+  defp assert_schema!(%{from: %Ecto.Query.FromExpr{source: {_source, schema}}}) when schema != nil, do: schema
   defp assert_schema!(query) do
     raise Ecto.QueryError, query: query, message: "expected a from expression with a schema"
   end
diff --git a/lib/ecto/query/builder.ex b/lib/ecto/query/builder.ex
index a6c33924..45173869 100644
--- a/lib/ecto/query/builder.ex
+++ b/lib/ecto/query/builder.ex
@@ -67,6 +67,24 @@ defmodule Ecto.Query.Builder do
     {expr, params_acc}
   end
 
+  def escape({:type, _, [{{:., _, [{var, _, context}, field]}, _, []} = expr, type]}, _type, params_take, vars, env)
+      when is_atom(var) and is_atom(context) and is_atom(field) do
+    escape_with_type(expr, type, params_take, vars, env)
+  end
+
+  def escape({:type, _, [{:fragment, _, [_ | _]} = expr, type]}, _type, params_take, vars, env) do
+    escape_with_type(expr, type, params_take, vars, env)
+  end
+
+  def escape({:type, _, [{:unsafe_fragment, _, [_ | _]} = expr, type]}, _type, params_take, vars, env) do
+    escape_with_type(expr, type, params_take, vars, env)
+  end
+
+  def escape({:type, _, [{agg, _, [_ | _]} = expr, type]}, _type, params_take, vars, env)
+      when agg in ~w(avg count max min sum)a do
+    escape_with_type(expr, type, params_take, vars, env)
+  end
+
   # fragments
   def escape({:fragment, _, [query]}, _type, params_acc, vars, env) when is_list(query) do
     {escaped, params_acc} =
@@ -337,6 +355,18 @@ defmodule Ecto.Query.Builder do
     error! "`#{Macro.to_string(other)}` is not a valid query expression"
   end
 
+  defp escape_with_type(expr, type, params_take, vars, env) do
+    type = validate_type!(type, vars)
+
+    {expr, params_take} = escape(expr, type, params_take, vars, {env, &escape_expansion/5})
+
+    {{:{}, [], [:type, [], [expr, type]]}, params_take}
+  end
+
+  defp escape_expansion(expr, type, params_take, vars, env) do
+    escape(expr, type, params_take, vars, env)
+  end
+
   defp wrap_nil(params, {:{}, _, [:^, _, [ix]]}) do
     Map.update!(params, ix, fn {val, type} ->
       quote do
diff --git a/lib/ecto/query/builder/from.ex b/lib/ecto/query/builder/from.ex
index bf36a229..6dc51004 100644
--- a/lib/ecto/query/builder/from.ex
+++ b/lib/ecto/query/builder/from.ex
@@ -79,7 +79,12 @@ defmodule Ecto.Query.Builder.From do
   end
 
   defp query(prefix, source, schema) do
-    {:%, [], [Ecto.Query, {:%{}, [], [from: {source, schema}, prefix: prefix, aliases: {:%{}, [], []}]}]}
+    {:%, [], [Ecto.Query,
+              {:%{}, [],
+               [from: {:%, [], [Ecto.Query.FromExpr,
+                                {:%{}, [], [source: {source, schema}]}]},
+                prefix: prefix,
+                aliases: {:%{}, [], []}]}]}
   end
 
   defp expand_from({left, right}, env) do
diff --git a/lib/ecto/query/builder/select.ex b/lib/ecto/query/builder/select.ex
index 7b9a46a0..b2dba769 100644
--- a/lib/ecto/query/builder/select.ex
+++ b/lib/ecto/query/builder/select.ex
@@ -96,24 +96,6 @@ defmodule Ecto.Query.Builder.Select do
     {expr, params_take}
   end
 
-  defp escape({:type, _, [{{:., _, [{var, _, context}, field]}, _, []} = expr, type]},
-              params_take, vars, env) when is_atom(var) and is_atom(context) and is_atom(field) do
-    escape_with_type(expr, type, params_take, vars, env)
-  end
-
-  defp escape({:type, _, [{:fragment, _, [_ | _]} = expr, type]}, params_take, vars, env) do
-    escape_with_type(expr, type, params_take, vars, env)
-  end
-
-  defp escape({:type, _, [{:unsafe_fragment, _, [_ | _]} = expr, type]}, params_take, vars, env) do
-    escape_with_type(expr, type, params_take, vars, env)
-  end
-
-  defp escape({:type, _, [{agg, _, [_ | _]} = expr, type]}, params_take, vars, env)
-       when agg in ~w(avg count max min sum)a do
-    escape_with_type(expr, type, params_take, vars, env)
-  end
-
   defp escape(expr, params_take, vars, env) do
     Builder.escape(expr, :any, params_take, vars, {env, &escape_expansion/5})
   end
@@ -122,13 +104,6 @@ defmodule Ecto.Query.Builder.Select do
     escape(expr, params_take, vars, env)
   end
 
-  defp escape_with_type(expr, type, params_take, vars, env) do
-    type = Builder.validate_type!(type, vars)
-    {expr, params_take} =
-      Builder.escape(expr, type, params_take, vars, {env, &escape_expansion/5})
-    {{:{}, [], [:type, [], [expr, type]]}, params_take}
-  end
-
   defp escape_pairs(pairs, params_take, vars, env) do
     Enum.map_reduce pairs, params_take, fn({k, v}, acc) ->
       {k, acc} = escape_key(k, acc, vars, env)
diff --git a/lib/ecto/query/inspect.ex b/lib/ecto/query/inspect.ex
index 0c4c6e94..9e54e1ed 100644
--- a/lib/ecto/query/inspect.ex
+++ b/lib/ecto/query/inspect.ex
@@ -68,20 +68,16 @@ defimpl Inspect, for: Ecto.Query do
                  limit, offset, lock, distinct, updates, select, preloads, assocs]
   end
 
-  defp bound_from(from, name), do: ["from #{name} in #{unbound_from from}"]
-
-  defp unbound_from(nil),           do: "query"
-  defp unbound_from({source, nil}), do: inspect source
-  defp unbound_from({nil, schema}),  do: inspect schema
-  defp unbound_from(from = {source, schema}) do
+  defp bound_from(nil, name), do: ["from #{name} in query"]
+  defp bound_from(%{source: source}, name), do: ["from #{name} in #{inspect_source source}"]
+
+  defp inspect_source(%Ecto.Query{} = query), do: "^" <> inspect(query)
+  defp inspect_source(%Ecto.SubQuery{query: query}), do: "subquery(#{to_string query})"
+  defp inspect_source({source, nil}), do: inspect source
+  defp inspect_source({nil, schema}), do: inspect schema
+  defp inspect_source({source, schema} = from) do
     inspect if source == schema.__schema__(:source), do: schema, else: from
   end
-  defp unbound_from(%Ecto.SubQuery{query: query}) do
-    "subquery(#{to_string query})"
-  end
-  defp unbound_from(%Ecto.Query{} = query) do
-    "^" <> inspect(query)
-  end
 
   defp joins(joins, names) do
     joins
@@ -100,7 +96,7 @@ defimpl Inspect, for: Ecto.Query do
   end
 
   defp join(%JoinExpr{qual: qual, source: source, on: on, as: as}, name, names) do
-    string = "#{name} in #{unbound_from source}"
+    string = "#{name} in #{inspect_source source}"
     [{join_qual(qual), string}] ++ kw_inspect(:as, as) ++ [on: expr(on, names)]
   end
 
@@ -237,11 +233,15 @@ defimpl Inspect, for: Ecto.Query do
   defp join_qual(:full),          do: :full_join
   defp join_qual(:cross),         do: :cross_join
 
-  defp collect_sources(query) do
-    [from_sources(query.from) | join_sources(query.joins)]
+  defp collect_sources(%{from: nil, joins: joins}) do
+    ["query" | join_sources(joins)]
+  end
+
+  defp collect_sources(%{from: %{source: source}, joins: joins}) do
+    [from_sources(source) | join_sources(joins)]
   end
 
-  defp from_sources(%Ecto.SubQuery{query: query}), do: from_sources(query.from)
+  defp from_sources(%Ecto.SubQuery{query: query}), do: from_sources(query.from.source)
   defp from_sources({source, schema}), do: schema || source
   defp from_sources(nil), do: "query"
 
@@ -252,7 +252,7 @@ defimpl Inspect, for: Ecto.Query do
       %JoinExpr{source: {:fragment, _, _}} ->
         "fragment"
       %JoinExpr{source: %Ecto.Query{from: from}} ->
-        from_sources(from)
+        from_sources(from.source)
       %JoinExpr{source: source} ->
         from_sources(source)
     end)
diff --git a/lib/ecto/query/planner.ex b/lib/ecto/query/planner.ex
index 4b805d9a..abc4581e 100644
--- a/lib/ecto/query/planner.ex
+++ b/lib/ecto/query/planner.ex
@@ -14,7 +14,7 @@ defmodule Ecto.Query.Planner do
   The from is moved as last join with the where conditions as its "on"
   in order to keep proper binding order.
   """
-  def query_to_joins(qual, %{from: from, wheres: wheres, joins: joins}, position) do
+  def query_to_joins(qual, source, %{wheres: wheres, joins: joins}, position) do
     on = %QueryExpr{file: __ENV__.file, line: __ENV__.line, expr: true, params: []}
 
     on =
@@ -22,7 +22,7 @@ defmodule Ecto.Query.Planner do
         merge_expr_and_params(op, acc, expr, params)
       end)
 
-    join = %JoinExpr{qual: qual, source: from, file: __ENV__.file, line: __ENV__.line, on: on}
+    join = %JoinExpr{qual: qual, source: source, file: __ENV__.file, line: __ENV__.line, on: on}
     last = length(joins) + position
 
     mapping = fn
@@ -224,14 +224,22 @@ defmodule Ecto.Query.Planner do
   @doc """
   Prepare all sources, by traversing and expanding joins.
   """
-  def prepare_sources(%{from: from} = query, adapter) do
-    from = from || error!(query, "query must have a from expression")
-    from = prepare_source(query, from, adapter)
-    {joins, sources, tail_sources} = prepare_joins(query, [from], length(query.joins), adapter)
+  def prepare_sources(query, adapter) do
+    {from, sources} = prepare_from(query, adapter)
+    {joins, sources, tail_sources} = prepare_joins(query, sources, length(query.joins), adapter)
     %{query | from: from, joins: joins |> Enum.reverse,
               sources: (tail_sources ++ sources) |> Enum.reverse |> List.to_tuple()}
   end
 
+  defp prepare_from(%{from: nil} = query, _adapter) do
+    error!(query, "query must have a from expression")
+  end
+
+  defp prepare_from(%{from: from} = query, adapter) do
+    source = prepare_source(query, from.source, adapter)
+    {%{from | source: source}, [source]}
+  end
+
   defp prepare_source(query, %Ecto.SubQuery{query: inner_query} = subquery, adapter) do
     try do
       {inner_query, params, key} = prepare(inner_query, :all, adapter, 0)
@@ -390,7 +398,7 @@ defmodule Ecto.Query.Planner do
     source_ix = counter
 
     {child_joins, child_sources, child_tail} =
-      prepare_joins(child, [child.from], offset + last_ix - 1, adapter)
+      prepare_joins(child, [child.from.source], offset + last_ix - 1, adapter)
 
     # Rewrite joins indexes as mentioned above
     child_joins = Enum.map(child_joins, &rewrite_join(&1, qual, ix, last_ix, source_ix, offset))
@@ -405,13 +413,13 @@ defmodule Ecto.Query.Planner do
                   child_sources ++ tail_sources, counter + 1, offset + length(child_sources), adapter)
   end
 
-  defp prepare_joins([%JoinExpr{source: %Ecto.Query{from: source} = join_query, qual: qual, on: on} = join|t],
+  defp prepare_joins([%JoinExpr{source: %Ecto.Query{} = join_query, qual: qual, on: on} = join|t],
                       query, joins, sources, tail_sources, counter, offset, adapter) do
     case join_query do
       %{order_bys: [], limit: nil, offset: nil, group_bys: [], joins: [],
         havings: [], preloads: [], assocs: [], distinct: nil, lock: nil} ->
-        source = prepare_source(query, source, adapter)
-        [join] = attach_on(query_to_joins(qual, %{join_query | from: source}, counter), on)
+        source = prepare_source(join_query, join_query.from.source, adapter)
+        [join] = attach_on(query_to_joins(qual, source, join_query, counter), on)
         prepare_joins(t, query, [join|joins], [source|sources], tail_sources, counter + 1, offset, adapter)
       _ ->
         error! query, join, "queries in joins can only have `where` conditions"
@@ -485,8 +493,8 @@ defmodule Ecto.Query.Planner do
     {query, Enum.reverse(params), finalize_cache(query, operation, cache, counter)}
   end
 
-  defp merge_cache(:from, _query, expr, {cache, params}, _adapter) do
-    {key, params} = source_cache(expr, params)
+  defp merge_cache(:from, _query, %{source: source}, {cache, params}, _adapter) do
+    {key, params} = source_cache(source, params)
     {merge_cache(key, cache, key != :nocache), params}
   end
 
@@ -710,11 +718,12 @@ defmodule Ecto.Query.Planner do
                    &validate_and_increment(&1, &2, &3, &4, operation, adapter))
   end
 
-  defp validate_and_increment(:from, query, %Ecto.SubQuery{}, _counter, kind, _adapter) when kind != :all do
+  defp validate_and_increment(:from, query, %{source: %Ecto.SubQuery{}}, _counter, kind, _adapter) when kind != :all do
     error! query, "`#{kind}` does not allow subqueries in `from`"
   end
-  defp validate_and_increment(:from, query, expr, counter, _kind, adapter) do
-    prewalk_source(expr, :from, query, expr, counter, adapter)
+  defp validate_and_increment(:from, query, %{source: source} = expr, counter, _kind, adapter) do
+    {source, acc} = prewalk_source(source, :from, query, expr, counter, adapter)
+    {%{expr | source: source}, acc}
   end
 
   defp validate_and_increment(kind, query, expr, counter, _operation, adapter)
diff --git a/lib/ecto/queryable.ex b/lib/ecto/queryable.ex
index defca806..3afd65bd 100644
--- a/lib/ecto/queryable.ex
+++ b/lib/ecto/queryable.ex
@@ -14,12 +14,15 @@ defimpl Ecto.Queryable, for: Ecto.Query do
 end
 
 defimpl Ecto.Queryable, for: Ecto.SubQuery do
-  def to_query(subquery), do: %Ecto.Query{from: subquery}
+  def to_query(subquery) do
+    %Ecto.Query{from: %Ecto.Query.FromExpr{source: subquery}}
+  end
 end
 
 defimpl Ecto.Queryable, for: BitString do
-  def to_query(source) when is_binary(source),
-    do: %Ecto.Query{from: {source, nil}}
+  def to_query(source) when is_binary(source) do
+    %Ecto.Query{from: %Ecto.Query.FromExpr{source: {source, nil}}}
+  end
 end
 
 defimpl Ecto.Queryable, for: Atom do
@@ -41,9 +44,10 @@ defimpl Ecto.Queryable, for: Atom do
 end
 
 defimpl Ecto.Queryable, for: Tuple do
-  def to_query({source, %Ecto.Query{from: {_, schema}} = query}) when is_binary(source),
-    do: %{query | from: {source, schema}}
-
-  def to_query({source, schema} = from) when is_binary(source) and is_atom(schema) and not is_nil(schema),
-    do: %Ecto.Query{from: from, prefix: schema.__schema__(:prefix)}
+  def to_query({source, schema} = from) when is_binary(source) and is_atom(schema) and not is_nil(schema) do
+    %Ecto.Query{
+      from: %Ecto.Query.FromExpr{source: from},
+      prefix: schema.__schema__(:prefix)
+    }
+  end
 end
diff --git a/lib/ecto/repo/queryable.ex b/lib/ecto/repo/queryable.ex
index 4be0b748..4f5be9fd 100644
--- a/lib/ecto/repo/queryable.ex
+++ b/lib/ecto/repo/queryable.ex
@@ -413,7 +413,7 @@ defmodule Ecto.Repo.Queryable do
     {{:., [], [{:&, [], [ix]}, field]}, [], []}
   end
 
-  defp assert_schema!(%{from: {_source, schema}}) when schema != nil, do: schema
+  defp assert_schema!(%{from: %{source: {_source, schema}}}) when schema != nil, do: schema
   defp assert_schema!(query) do
     raise Ecto.QueryError,
       query: query,
diff --git a/lib/ecto/repo/schema.ex b/lib/ecto/repo/schema.ex
index 8ae41dd1..1d0281b4 100644
--- a/lib/ecto/repo/schema.ex
+++ b/lib/ecto/repo/schema.ex
@@ -528,7 +528,7 @@ defmodule Ecto.Repo.Schema do
       %{query | prefix: prefix}
       |> Ecto.Query.Planner.prepare(:update_all, adapter, counter)
 
-    unless query.from == from do
+    unless query.from.source == from do
       raise ArgumentError, "cannot run on_conflict: query because the query " <>
                            "has a different {source, schema} pair than the " <>
                            "original struct/changeset/query. Got #{inspect query.from} " <>
diff --git a/lib/ecto/schema.ex b/lib/ecto/schema.ex
index f0f7a14a..1fbe3afb 100644
--- a/lib/ecto/schema.ex
+++ b/lib/ecto/schema.ex
@@ -1912,12 +1912,18 @@ defmodule Ecto.Schema do
     hash = :erlang.phash2({primary_key, fields})
 
     quote do
-      def __schema__(:query),       do: %Ecto.Query{from: {unquote(source), __MODULE__}, prefix: unquote(prefix)}
       def __schema__(:prefix),      do: unquote(prefix)
       def __schema__(:source),      do: unquote(source)
       def __schema__(:fields),      do: unquote(field_names)
       def __schema__(:primary_key), do: unquote(primary_key)
       def __schema__(:hash),        do: unquote(hash)
+
+      def __schema__(:query) do
+        %Ecto.Query{
+          from: %Ecto.Query.FromExpr{source: {unquote(source), __MODULE__}},
+          prefix: unquote(prefix)
+        }
+      end
     end
   end
 


Test patch:
diff --git a/test/ecto/adapters/mysql_test.exs b/test/ecto/adapters/mysql_test.exs
index 6a29c811..297c722d 100644
--- a/test/ecto/adapters/mysql_test.exs
+++ b/test/ecto/adapters/mysql_test.exs
@@ -41,6 +41,9 @@ defmodule Ecto.Adapters.MySQLTest do
 
     schema "schema3" do
       field :binary, :binary
+      embeds_one :map do
+        field :value, :decimal
+      end
     end
   end
 
@@ -222,6 +225,17 @@ defmodule Ecto.Adapters.MySQLTest do
     assert all(query) == ~s{SELECT NOT (s0.`x` IS NULL) FROM `schema` AS s0}
   end
 
+  test "order_by and types" do
+    query =
+      normalize from(e in "schema3",
+        order_by: type(fragment("(?->>?)", field(e, ^:map), ^"value"), ^:decimal),
+        select: true)
+
+    result = "SELECT TRUE FROM `schema3` AS s0 ORDER BY (s0.`map`->>?) + 0"
+
+    assert all(query) == result
+  end
+
   test "fragments" do
     query = Schema |> select([r], fragment("lcase(?)", r.x)) |> normalize
     assert all(query) == ~s{SELECT lcase(s0.`x`) FROM `schema` AS s0}
diff --git a/test/ecto/adapters/postgres_test.exs b/test/ecto/adapters/postgres_test.exs
index a3ce9e19..3abcbcc4 100644
--- a/test/ecto/adapters/postgres_test.exs
+++ b/test/ecto/adapters/postgres_test.exs
@@ -43,6 +43,9 @@ defmodule Ecto.Adapters.PostgresTest do
       field :list1, {:array, :string}
       field :list2, {:array, :integer}
       field :binary, :binary
+      embeds_one :map do
+        field :value, :decimal
+      end
     end
   end
 
@@ -395,6 +398,17 @@ defmodule Ecto.Adapters.PostgresTest do
     assert all(query) == String.trim(result)
   end
 
+  test "order_by and types" do
+    query =
+      normalize from(e in "schema3",
+        order_by: type(fragment("(?->>?)", field(e, ^:map), ^"value"), ^:decimal),
+        select: true)
+
+    result = "SELECT TRUE FROM \"schema3\" AS s0 ORDER BY (s0.\"map\"->>$1)::decimal"
+
+    assert all(query) == result
+  end
+
   test "fragments and types" do
     query =
       normalize from(e in "schema",
diff --git a/test/ecto/query/builder/join_test.exs b/test/ecto/query/builder/join_test.exs
index ff8ab3be..d6744991 100644
--- a/test/ecto/query/builder/join_test.exs
+++ b/test/ecto/query/builder/join_test.exs
@@ -44,7 +44,7 @@ defmodule Ecto.Query.Builder.JoinTest do
 
     qual = :inner
     source = from c in "comments", where: c.public
-    assert %{joins: [%{source: %Ecto.Query{from: {"comments", nil}}}]} =
+    assert %{joins: [%{source: %Ecto.Query{from: %{source: {"comments", nil}}}}]} =
             join("posts", qual, [p], c in ^source, on: true)
   end
 
diff --git a/test/ecto/query/subquery_test.exs b/test/ecto/query/subquery_test.exs
index 789372ce..5babe247 100644
--- a/test/ecto/query/subquery_test.exs
+++ b/test/ecto/query/subquery_test.exs
@@ -55,14 +55,14 @@ defmodule Ecto.Query.SubqueryTest do
 
   test "prepare: subqueries" do
     {query, params, key} = prepare(from(subquery(Post), []))
-    assert %{query: %Ecto.Query{}, params: []} = query.from
+    assert %{query: %Ecto.Query{}, params: []} = query.from.source
     assert params == []
     assert key == [:all, 0, [:all, 0, {"posts", Post, 127044068}]]
 
     posts = from(p in Post, where: p.title == ^"hello")
     query = from(c in Comment, join: p in subquery(posts), on: c.post_id == p.id)
     {query, params, key} = prepare(query, [])
-    assert {"comments", Comment} = query.from
+    assert {"comments", Comment} = query.from.source
     assert [%{source: %{query: %Ecto.Query{}, params: ["hello"]}}] = query.joins
     assert params == ["hello"]
     assert [[], 0, {:join, [{:inner, [:all|_], _}]}, {"comments", _, _}] = key
@@ -104,33 +104,33 @@ defmodule Ecto.Query.SubqueryTest do
       query = prepare(from(subquery(Post), [])) |> elem(0)
       assert "%Ecto.Query.SubqueryTest.Post{id: &0.id(), title: &0.title(), " <>
              "text: &0.text()}" =
-             Macro.to_string(query.from.query.select.expr)
+             Macro.to_string(query.from.source.query.select.expr)
     end
 
     test "supports field selector" do
       query = from p in "posts", select: p.text
       query = prepare(from(subquery(query), [])) |> elem(0)
       assert "%{text: &0.text()}" =
-             Macro.to_string(query.from.query.select.expr)
+             Macro.to_string(query.from.source.query.select.expr)
 
       query = from p in Post, select: p.text
       query = prepare(from(subquery(query), [])) |> elem(0)
       assert "%{text: &0.text()}" =
-             Macro.to_string(query.from.query.select.expr)
+             Macro.to_string(query.from.source.query.select.expr)
     end
 
     test "supports maps" do
       query = from p in Post, select: %{text: p.text}
       query = prepare(from(subquery(query), [])) |> elem(0)
       assert "%{text: &0.text()}" =
-             Macro.to_string(query.from.query.select.expr)
+             Macro.to_string(query.from.source.query.select.expr)
     end
 
     test "supports structs" do
       query = from p in Post, select: %Post{text: p.text}
       query = prepare(from(subquery(query), [])) |> elem(0)
       assert "%Ecto.Query.SubqueryTest.Post{text: &0.text()}" =
-             Macro.to_string(query.from.query.select.expr)
+             Macro.to_string(query.from.source.query.select.expr)
     end
 
     test "supports update in maps" do
@@ -138,7 +138,7 @@ defmodule Ecto.Query.SubqueryTest do
       query = prepare(from(subquery(query), [])) |> elem(0)
       assert "%Ecto.Query.SubqueryTest.Post{id: &0.id(), title: &0.title(), " <>
              "text: &0.title()}" =
-             Macro.to_string(query.from.query.select.expr)
+             Macro.to_string(query.from.source.query.select.expr)
 
       query = from p in Post, select: %{p | unknown: p.title}
       assert_raise Ecto.SubQueryError, ~r/invalid key `:unknown` on map update in subquery/, fn ->
@@ -151,11 +151,11 @@ defmodule Ecto.Query.SubqueryTest do
       query = prepare(from(subquery(query), [])) |> elem(0)
       assert "%Ecto.Query.SubqueryTest.Post{id: &0.id(), title: &0.title(), " <>
              "text: &0.title()}" =
-             Macro.to_string(query.from.query.select.expr)
+             Macro.to_string(query.from.source.query.select.expr)
 
       query = from p in Post, select: merge(%{}, %{})
       query = prepare(from(subquery(query), [])) |> elem(0)
-      assert "%{}" = Macro.to_string(query.from.query.select.expr)
+      assert "%{}" = Macro.to_string(query.from.source.query.select.expr)
     end
 
     test "requires atom keys for maps" do
@@ -226,8 +226,8 @@ defmodule Ecto.Query.SubqueryTest do
 
     {query, params} = normalize_with_params(query)
     assert [_, {:^, _, [0]}] = query.select.expr
-    assert [%{expr: {:==, [], [_, {:^, [], [1]}]}}] = query.from.query.wheres
-    assert [%{expr: [asc: {:==, [], [_, {:^, [], [2]}]}]}] = query.from.query.order_bys
+    assert [%{expr: {:==, [], [_, {:^, [], [1]}]}}] = query.from.source.query.wheres
+    assert [%{expr: [asc: {:==, [], [_, {:^, [], [2]}]}]}] = query.from.source.query.order_bys
     assert [%{expr: {:==, [], [_, {:^, [], [3]}]}}] = query.wheres
     assert params == ["first", "hello", "world", "last"]
   end
diff --git a/test/ecto/query_test.exs b/test/ecto/query_test.exs
index 83b4cbfd..8a160dca 100644
--- a/test/ecto/query_test.exs
+++ b/test/ecto/query_test.exs
@@ -76,19 +76,13 @@ defmodule Ecto.QueryTest do
         quote_and_eval(from("posts", 123))
       end
     end
-
-    test "can override the source for existing queries" do
-      query = %Query{from: {"posts", nil}}
-      query = from q in {"new_posts", query}, where: true
-      assert query.from == {"new_posts", nil}
-    end
   end
 
   describe "subqueries" do
     test "builds a subquery struct" do
-      assert subquery("posts").query.from == {"posts", nil}
-      assert subquery(subquery("posts")).query.from == {"posts", nil}
-      assert subquery(subquery("posts").query).query.from == {"posts", nil}
+      assert subquery("posts").query.from.source == {"posts", nil}
+      assert subquery(subquery("posts")).query.from.source == {"posts", nil}
+      assert subquery(subquery("posts").query).query.from.source == {"posts", nil}
     end
 
     test "prefix is not applied if left blank" do
diff --git a/test/support/test_repo.exs b/test/support/test_repo.exs
index 83db311f..b5c6f5d4 100644
--- a/test/support/test_repo.exs
+++ b/test/support/test_repo.exs
@@ -45,12 +45,12 @@ defmodule Ecto.TestAdapter do
     Process.get(:test_repo_all_results, {1, [[]]})
   end
 
-  def execute(_repo, _meta, {:nocache, {:delete_all, %{from: {_, SchemaMigration}}}}, [version], _) do
+  def execute(_repo, _meta, {:nocache, {:delete_all, %{from: %{source: {_, SchemaMigration}}}}}, [version], _) do
     Process.put(:migrated_versions, List.delete(migrated_versions(), version))
     {1, nil}
   end
 
-  def execute(_repo, meta, {:nocache, {op, %{from: {source, _}}}}, _params, _opts) do
+  def execute(_repo, meta, {:nocache, {op, %{from: %{source: {source, _}}}}}, _params, _opts) do
     send test_process(), {op, {meta.prefix, source}}
     {1, nil}
   end