Repository: elixir-ecto/ecto
Pull request #2155 (2017-08-04 19:15:43)

Introduce an AST for preprocessing + postprocessing
This reduces the complexity in the adapters, reduces
the caching cost per entry and speeds up deserialization.

josevalim: Hi @shdblowers, @jbachhardie, @AndrewDryga!

If this pull request is merged, it will require changes in adapters in order to support v2.2. The good news are: those changes can be done in a way to remain compatible with earlier codes and they all simplify the adapter code!
ebert[bot]: Ebert has finished reviewing this Pull Request and has found:

* 4 possible new issues (including those that may have been commented here).
* 5 fixed issues! 🎉

You can see more details about this review at https://ebertapp.io/github/elixir-ecto/ecto/pulls/2155.

<!-- Ebert auto generated comment -->


Patch:
diff --git a/lib/ecto/adapter.ex b/lib/ecto/adapter.ex
index b7404479..424d0a5a 100644
--- a/lib/ecto/adapter.ex
+++ b/lib/ecto/adapter.ex
@@ -7,8 +7,7 @@ defmodule Ecto.Adapter do
   @type t :: module
 
   @typedoc "Ecto.Query metadata fields (stored in cache)"
-  @type query_meta :: %{prefix: binary | nil, sources: tuple, assocs: term,
-                        preloads: term, select: term, fields: [term]}
+  @type query_meta :: %{prefix: binary | nil, sources: tuple, preloads: term, select: map}
 
   @typedoc "Ecto.Schema metadata fields"
   @type schema_meta :: %{source: source, schema: atom, context: term, autogenerate_id: {atom, :id | :binary_id}}
@@ -20,7 +19,7 @@ defmodule Ecto.Adapter do
   @type returning :: [atom]
   @type prepared :: term
   @type cached :: term
-  @type process :: (field :: Macro.t, value :: term, context :: term -> term)
+  @type process :: (term -> term)
   @type autogenerate_id :: {field :: atom, type :: :id | :binary_id, value :: term} | nil
   @type on_conflict :: {:raise, list(), []} |
                        {:nothing, list(), [atom]} |
diff --git a/lib/ecto/adapters/mysql/connection.ex b/lib/ecto/adapters/mysql/connection.ex
index 3be70a42..c55dde1f 100644
--- a/lib/ecto/adapters/mysql/connection.ex
+++ b/lib/ecto/adapters/mysql/connection.ex
@@ -338,13 +338,10 @@ if Code.ensure_loaded?(Mariaex) do
       [name, ?. | quote_name(field)]
     end
 
-    defp expr({:&, _, [idx, fields, _counter]}, sources, query) do
-      {source, name, schema} = elem(sources, idx)
-      if is_nil(schema) and is_nil(fields) do
-        error!(query, "MySQL does not support selecting all fields from #{source} without a schema. " <>
-                      "Please specify a schema or specify exactly which fields you want to select")
-      end
-      intersperse_map(fields, ", ", &[name, ?. | quote_name(&1)])
+    defp expr({:&, _, [idx]}, sources, query) do
+      {source, _name, _schema} = elem(sources, idx)
+      error!(query, "MySQL does not support selecting all fields from #{source} without a schema. " <>
+                    "Please specify a schema or specify exactly which fields you want to select")
     end
 
     defp expr({:in, _, [_left, []]}, _sources, _query) do
@@ -377,8 +374,8 @@ if Code.ensure_loaded?(Mariaex) do
       ["NOT (", expr(expr, sources, query), ?)]
     end
 
-    defp expr(%Ecto.SubQuery{query: query, fields: fields}, _sources, _query) do
-      query.select.fields |> put_in(fields) |> all()
+    defp expr(%Ecto.SubQuery{query: query}, _sources, _query) do
+      all(query)
     end
 
     defp expr({:fragment, _, [kw]}, _sources, query) when is_list(kw) or tuple_size(kw) == 3 do
diff --git a/lib/ecto/adapters/postgres/connection.ex b/lib/ecto/adapters/postgres/connection.ex
index a00594b3..7486b018 100644
--- a/lib/ecto/adapters/postgres/connection.ex
+++ b/lib/ecto/adapters/postgres/connection.ex
@@ -418,13 +418,10 @@ if Code.ensure_loaded?(Postgrex) do
       quote_qualified_name(field, sources, idx)
     end
 
-    defp expr({:&, _, [idx, fields, _counter]}, sources, query) do
-      {source, name, schema} = elem(sources, idx)
-      if is_nil(schema) and is_nil(fields) do
-        error!(query, "PostgreSQL does not support selecting all fields from #{source} without a schema. " <>
-                      "Please specify a schema or specify exactly which fields you want to select")
-      end
-      intersperse_map(fields, ", ", &[name, ?. | quote_name(&1)])
+    defp expr({:&, _, [idx]}, sources, query) do
+      {source, _name, _schema} = elem(sources, idx)
+      error!(query, "PostgreSQL does not support selecting all fields from #{source} without a schema. " <>
+                    "Please specify a schema or specify exactly which fields you want to select")
     end
 
     defp expr({:in, _, [_left, []]}, _sources, _query) do
@@ -452,8 +449,8 @@ if Code.ensure_loaded?(Postgrex) do
       ["NOT (", expr(expr, sources, query), ?)]
     end
 
-    defp expr(%Ecto.SubQuery{query: query, fields: fields}, _sources, _query) do
-      query.select.fields |> put_in(fields) |> all()
+    defp expr(%Ecto.SubQuery{query: query}, _sources, _query) do
+      all(query)
     end
 
     defp expr({:fragment, _, [kw]}, _sources, query) when is_list(kw) or tuple_size(kw) == 3 do
diff --git a/lib/ecto/adapters/sql.ex b/lib/ecto/adapters/sql.ex
index 621b5fbe..31348099 100644
--- a/lib/ecto/adapters/sql.ex
+++ b/lib/ecto/adapters/sql.ex
@@ -388,32 +388,15 @@ defmodule Ecto.Adapters.SQL do
     do_execute(repo, meta, prepared, params, mapper, put_source(opts, meta))
   end
 
-  defp do_execute(repo, _meta, {:cache, update, {id, prepared}}, params, nil, opts) do
-    execute_and_cache(repo, id, update, prepared, params, nil, opts)
-  end
-
-  defp do_execute(repo, %{fields: fields, sources: sources}, {:cache, update, {id, prepared}}, params, process, opts) do
-    mapper = &process_row(&1, process, fields, sources)
+  defp do_execute(repo, _meta, {:cache, update, {id, prepared}}, params, mapper, opts) do
     execute_and_cache(repo, id, update, prepared, params, mapper, opts)
   end
 
-  defp do_execute(repo, _meta, {:cached, reset, {id, cached}}, params, nil, opts) do
-    execute_or_reset(repo, id, reset, cached, params, nil, opts)
-  end
-
-  defp do_execute(repo, %{fields: fields, sources: sources}, {:cached, reset, {id, cached}}, params, process, opts) do
-    mapper = &process_row(&1, process, fields, sources)
+  defp do_execute(repo, _meta, {:cached, reset, {id, cached}}, params, mapper, opts) do
     execute_or_reset(repo, id, reset, cached, params, mapper, opts)
   end
 
-  defp do_execute(repo, _meta, {:nocache, {_id, prepared}}, params, nil, opts) do
-    %{rows: rows, num_rows: num} =
-      sql_call!(repo, :execute, [prepared], params, nil, opts)
-    {num, rows}
-  end
-
-  defp do_execute(repo, %{fields: fields, sources: sources}, {:nocache, {_id, prepared}}, params, process, opts) do
-    mapper = &process_row(&1, process, fields, sources)
+  defp do_execute(repo, _meta, {:nocache, {_id, prepared}}, params, mapper, opts) do
     %{rows: rows, num_rows: num} =
       sql_call!(repo, :execute, [prepared], params, mapper, opts)
     {num, rows}
@@ -491,30 +474,15 @@ defmodule Ecto.Adapters.SQL do
     do_stream(repo, meta, prepared, params, mapper, put_source(opts, meta))
   end
 
-  def do_stream(repo, _meta, {:cache, _, {_, prepared}}, params, nil, opts) do
-    prepare_stream(repo, prepared, params, nil, opts)
-  end
-
-  def do_stream(repo, %{fields: fields, sources: sources}, {:cache, _, {_, prepared}}, params, process, opts) do
-    mapper = &process_row(&1, process, fields, sources)
+  def do_stream(repo, _meta, {:cache, _, {_, prepared}}, params, mapper, opts) do
     prepare_stream(repo, prepared, params, mapper, opts)
   end
 
-  def do_stream(repo, _, {:cached, _, {_, cached}}, params, nil, opts) do
-    prepare_stream(repo, String.Chars.to_string(cached), params, nil, opts)
-  end
-
-  def do_stream(repo, %{fields: fields, sources: sources}, {:cached, _, {_, cached}}, params, process, opts) do
-    mapper = &process_row(&1, process, fields, sources)
+  def do_stream(repo, _, {:cached, _, {_, cached}}, params, mapper, opts) do
     prepare_stream(repo, String.Chars.to_string(cached), params, mapper, opts)
   end
 
-  def do_stream(repo, _meta, {:nocache, {_id, prepared}}, params, nil, opts) do
-    prepare_stream(repo, prepared, params, nil, opts)
-  end
-
-  def do_stream(repo, %{fields: fields, sources: sources}, {:nocache, {_id, prepared}}, params, process, opts) do
-    mapper = &process_row(&1, process, fields, sources)
+  def do_stream(repo, _meta, {:nocache, {_id, prepared}}, params, mapper, opts) do
     prepare_stream(repo, prepared, params, mapper, opts)
   end
 
@@ -570,33 +538,6 @@ defmodule Ecto.Adapters.SQL do
     end
   end
 
-  defp process_row(row, process, fields, sources) do
-    num_sources = tuple_size(sources)
-    Enum.map_reduce(fields, row, fn
-      {:&, _, [_, _, counter]} = field, acc when num_sources == 1 ->
-        {val, rest} = Enum.split(acc, counter)
-        {process.(field, val, nil), rest}
-      {:&, _, [_, _, counter]} = field, acc ->
-        case split_and_not_nil(acc, counter, true, []) do
-          {nil, rest} -> {nil, rest}
-          {val, rest} -> {process.(field, val, nil), rest}
-        end
-      field, [h|t] ->
-        {process.(field, h, nil), t}
-    end) |> elem(0)
-  end
-
-  defp split_and_not_nil(rest, 0, true, _acc), do: {nil, rest}
-  defp split_and_not_nil(rest, 0, false, acc), do: {:lists.reverse(acc), rest}
-
-  defp split_and_not_nil([nil|t], count, all_nil?, acc) do
-    split_and_not_nil(t, count - 1, all_nil?, [nil|acc])
-  end
-
-  defp split_and_not_nil([h|t], count, _all_nil?, acc) do
-    split_and_not_nil(t, count - 1, false, [h|acc])
-  end
-
   ## Transactions
 
   @doc false
diff --git a/lib/ecto/query.ex b/lib/ecto/query.ex
index 5d71cfcf..f439206a 100644
--- a/lib/ecto/query.ex
+++ b/lib/ecto/query.ex
@@ -2,7 +2,7 @@ defmodule Ecto.SubQuery do
   @moduledoc """
   Stores subquery information.
   """
-  defstruct [:query, :params, :fields, :meta, :cache]
+  defstruct [:query, :params, :select, :cache]
 end
 
 defmodule Ecto.Query do
diff --git a/lib/ecto/query/planner.ex b/lib/ecto/query/planner.ex
index 6710c3a2..4773056a 100644
--- a/lib/ecto/query/planner.ex
+++ b/lib/ecto/query/planner.ex
@@ -172,33 +172,13 @@ defmodule Ecto.Query.Planner do
   end
 
   defp query_without_cache(query, operation, adapter, counter) do
-    %{select: select} = query = normalize(query, operation, adapter, counter)
+    {query, select} = normalize(query, operation, adapter, counter)
     {cache, prepared} = adapter.prepare(operation, query)
     {cache, select, prepared}
   end
 
-  defp build_meta(%{prefix: prefix, sources: sources, assocs: assocs, preloads: preloads},
-                  %{expr: select, fields: fields, take: take}) do
-    %{prefix: prefix, sources: build_sources_meta(sources, 1, tuple_size(sources)),
-      fields: fields, take: take, assocs: assocs, preloads: preloads, select: select}
-  end
-  defp build_meta(%{prefix: prefix, sources: sources, assocs: assocs, preloads: preloads},
-                  nil) do
-    %{prefix: prefix, sources: build_sources_meta(sources, 1, tuple_size(sources)),
-      fields: nil, take: nil, assocs: assocs, preloads: preloads, select: nil}
-  end
-
-  defp build_sources_meta(sources, pos, size) when pos <= size do
-    case :erlang.element(pos, sources) do
-      %Ecto.SubQuery{query: %{select: select} = query} = subquery ->
-        subquery = %{subquery | meta: build_meta(query, select), cache: nil, query: nil, params: nil}
-        build_sources_meta(:erlang.setelement(pos, sources, subquery), pos + 1, size)
-      _ ->
-        build_sources_meta(sources, pos + 1, size)
-    end
-  end
-  defp build_sources_meta(sources, _, _) do
-    sources
+  defp build_meta(%{prefix: prefix, sources: sources, preloads: preloads}, select) do
+    %{prefix: prefix, select: select, preloads: preloads, sources: sources}
   end
 
   @doc """
@@ -236,10 +216,9 @@ defmodule Ecto.Query.Planner do
   defp prepare_source(query, %Ecto.SubQuery{query: inner_query} = subquery, adapter) do
     try do
       {inner_query, params, key} = prepare(inner_query, :all, adapter, 0)
-      %{select: select} = inner_query = inner_query |> returning(true) |> normalize_select()
-      {select, fields} = subquery_fields(inner_query, select)
-      %{subquery | query: %{inner_query | select: select},
-                   params: params, cache: key, fields: fields}
+      assert_no_subquery_assocs!(inner_query)
+      {inner_query, select} = inner_query |> returning(true) |> subquery_select()
+      %{subquery | query: inner_query, params: params, cache: key, select: select}
     rescue
       e -> raise Ecto.SubQueryError, query: query, exception: e
     end
@@ -252,58 +231,80 @@ defmodule Ecto.Query.Planner do
   defp prepare_source(_query, {:fragment, _, _} = source, _adapter),
     do: source
 
-  defp subquery_fields(%{assocs: assocs, preloads: preloads} = query, _select)
-      when assocs != [] or preloads != [] do
+  defp assert_no_subquery_assocs!(%{assocs: assocs, preloads: preloads} = query)
+       when assocs != [] or preloads != [] do
     error!(query, "cannot preload associations in subquery")
   end
-  defp subquery_fields(query, %{expr: {:%{}, _, [{:|, _, [{:&, [], [ix]}, pairs]}]} = expr,
-                                fields: [{:&, meta, [ix, [_ | _] = fields, _]} | rest]} = select) do
-    # In case of map updates, we need to remove duplicated fields
-    # at query time because we use the field names as aliases and
-    # duplicate aliases will lead to invalid queries.
-    map_fields = subquery_map_fields(query, expr, pairs)
-    map_keys = Keyword.keys(map_fields)
-    kept_keys = fields -- map_keys
-    invalid_keys = map_keys -- fields
+  defp assert_no_subquery_assocs!(query) do
+    query
+  end
 
-    if invalid_keys != [] do
-      error!(query, "invalid key `#{inspect hd(invalid_keys)}` on map update in subquery")
+  defp subquery_select(%{select: %{expr: expr, take: take}} = query) do
+    expr =
+      case subquery_select(expr, take, query) do
+        {nil, fields} ->
+          {:%{}, [], fields}
+        {schema, fields} ->
+          {:%, [], [schema, {:%{}, [], fields}]}
+      end
+
+    {select, _fields, _from} =
+      collect_fields(expr, [], :error, query, take)
+
+    {put_in(query.select.expr, expr), select}
+  end
+
+  defp subquery_select({:%{}, _, [{:|, _, [{:&, [], [ix]}, pairs]}]} = expr, take, query) do
+    assert_subquery_fields!(query, expr, pairs)
+    {_, schema} = get_source!(:select, query, ix)
+    {_, fields} = Map.get_lazy(take, ix, fn -> {:any, schema.__schema__(:fields)} end)
+
+    update_keys = Keyword.keys(pairs)
+    case update_keys -- fields do
+      [] -> :ok
+      [key | _] -> error!(query, "invalid key `#{inspect key}` on map update in subquery")
     end
 
-    select_fields = [{:&, meta, [ix, kept_keys, length(kept_keys)]} | rest]
-    subquery_fields = subquery_selector_fields(kept_keys, ix) ++ map_fields
-    {%{select | fields: select_fields}, subquery_fields}
+    # In case of map updates, we need to remove duplicated fields
+    # at query time because we use the field names as aliases and
+    # duplicate aliases will lead to invalid queries.
+    kept_keys = fields -- update_keys
+    {schema, subquery_fields(kept_keys, ix) ++ pairs}
   end
-  defp subquery_fields(query, %{expr: {:%{}, _, pairs} = expr} = select) do
-    {select, subquery_map_fields(query, expr, pairs)}
+  defp subquery_select({:%{}, _, pairs} = expr, _take, query) do
+    assert_subquery_fields!(query, expr, pairs)
+    {nil, pairs}
   end
-  defp subquery_fields(_query, %{expr: {:&, _, [ix]},
-                                 fields: [{:&, _, [ix, [_ | _] = fields, _]}]} = select) do
-    {select, subquery_selector_fields(fields, ix)}
+  defp subquery_select({:&, _, [ix]}, _take, query) do
+    {_, schema} = get_source!(:subquery, query, ix)
+    fields = schema.__schema__(:fields)
+    {schema, subquery_fields(fields, ix)}
   end
-  defp subquery_fields(_query, %{expr: {{:., _, [{:&, _, [ix]}, field]}, _, []},
-                                 fields: [{{:., _, [{:&, _, [ix]}, field]}, _, []}]} = select) do
-    {select, [{field, {{:., [], [{:&, [], [ix]}, field]}, [], []}}]}
+  defp subquery_select({{:., _, [{:&, _, [ix]}, field]}, _, []}, _take, _query) do
+    {nil, subquery_fields([field], ix)}
   end
-  defp subquery_fields(query, %{expr: expr}) do
+  defp subquery_select(expr, _take, query) do
     error!(query, "subquery must select a source (t), a field (t.field) or a map, got: `#{Macro.to_string(expr)}`")
   end
 
-  defp subquery_selector_fields(fields, ix) do
-    Enum.map(fields, fn field ->
+  defp subquery_fields(fields, ix) do
+    for field <- fields, is_atom(field) do
       {field, {{:., [], [{:&, [], [ix]}, field]}, [], []}}
-    end)
+    end
   end
 
-  defp subquery_map_fields(query, expr, pairs) do
-    Enum.map(pairs, fn
+  defp subquery_types(%{select: {:map, types}}), do: types
+  defp subquery_types(%{select: {:struct, _name, types}}), do: types
+
+  defp assert_subquery_fields!(query, expr, pairs) do
+    Enum.each(pairs, fn
       {key, _} when not is_atom(key) ->
         error!(query, "only atom keys are allowed when selecting a map in subquery, got: `#{Macro.to_string(expr)}`")
       {key, value} ->
         if valid_subquery_value?(value) do
           {key, value}
         else
-          error!(query, "containers (maps, lists, tuples) or sources (t) are not allowed when selecting a map in subquery, got: `#{Macro.to_string(expr)}`")
+          error!(query, "maps, lists, tuples and sources are not allowed as map values in subquery, got: `#{Macro.to_string(expr)}`")
         end
     end)
   end
@@ -425,13 +426,11 @@ defmodule Ecto.Query.Planner do
                               "because it does not have a schema"
       {_, schema} ->
         schema
-      %Ecto.SubQuery{query: %{select: %{expr: {:&, _, [ix]}}, sources: sources}} when is_integer(ix) ->
-        schema_for_association_join!(query, join, elem(sources, ix))
-      %Ecto.SubQuery{query: %{select: %{expr: {:%{}, _, [{:|, _, [{:&, _, [ix]}, _fields]}]}}, sources: sources}} when is_integer(ix) ->
-        schema_for_association_join!(query, join, elem(sources, ix))
+      %Ecto.SubQuery{select: {:struct, schema, _}} ->
+        schema
       %Ecto.SubQuery{} ->
         error! query, join, "can only perform association joins on subqueries " <>
-                            "that return a single source in select"
+                            "that return a source with schema in select"
       _ ->
         error! query, join, "can only perform association joins on sources with a schema"
     end
@@ -688,7 +687,7 @@ defmodule Ecto.Query.Planner do
     error! query, "`#{kind}` does not allow subqueries in `from`"
   end
   defp validate_and_increment(:from, query, expr, counter, _kind, adapter) do
-    prewalk(expr, :from, query, expr, counter, adapter)
+    prewalk_source(expr, :from, query, expr, counter, adapter)
   end
 
   defp validate_and_increment(kind, query, expr, counter, _operation, adapter)
@@ -715,25 +714,35 @@ defmodule Ecto.Query.Planner do
 
   defp validate_and_increment(:join, query, exprs, counter, _operation, adapter) do
     Enum.map_reduce exprs, counter, fn join, acc ->
-      {source, acc} = prewalk(join.source, :join, query, join, acc, adapter)
+      {source, acc} = prewalk_source(join.source, :join, query, join, acc, adapter)
       {on, acc} = prewalk(:join, query, join.on, acc, adapter)
       {%{join | on: on, source: source, params: nil}, acc}
     end
   end
 
-  defp prewalk(kind, query, expr, counter, adapter) do
-    {inner, acc} = prewalk(expr.expr, kind, query, expr, counter, adapter)
-    {%{expr | expr: inner, params: nil}, acc}
+  defp prewalk_source({:fragment, meta, fragments}, kind, query, expr, acc, adapter) do
+    {fragments, acc} = prewalk(fragments, kind, query, expr, acc, adapter)
+    {{:fragment, meta, fragments}, acc}
   end
-
-  defp prewalk(%Ecto.SubQuery{query: inner_query} = subquery, _kind, query, _expr, counter, adapter) do
+  defp prewalk_source(%Ecto.SubQuery{query: inner_query} = subquery, _kind, query, _expr, counter, adapter) do
     try do
       {inner_query, counter} = normalize_query(inner_query, :all, adapter, counter)
-      {%{subquery | query: inner_query, params: nil}, counter}
+      {inner_query, _} = normalize_select(inner_query)
+      keys = subquery |> subquery_types() |> Keyword.keys()
+      inner_query = update_in(inner_query.select.fields, &Enum.zip(keys, &1))
+      {%{subquery | query: inner_query}, counter}
     rescue
       e -> raise Ecto.SubQueryError, query: query, exception: e
     end
   end
+  defp prewalk_source(source, _kind, _query, _expr, acc, _adapter) do
+    {source, acc}
+  end
+
+  defp prewalk(kind, query, expr, counter, adapter) do
+    {inner, acc} = prewalk(expr.expr, kind, query, expr, counter, adapter)
+    {%{expr | expr: inner, params: nil}, acc}
+  end
 
   defp prewalk({:in, in_meta, [left, {:^, meta, [param]}]}, kind, query, expr, acc, adapter) do
     {left, acc} = prewalk(left, kind, query, expr, acc, adapter)
@@ -804,149 +813,220 @@ defmodule Ecto.Query.Planner do
     {{:^, meta, [acc, length]}, acc + 1}
   end
 
-  defp normalize_select(%{select: %{fields: nil} = select} = query) do
-    %{query | select: normalize_fields(query, select)}
+  defp normalize_select(%{select: nil} = query) do
+    {query, nil}
   end
   defp normalize_select(query) do
-    query
-  end
+    %{assocs: assocs, preloads: preloads, select: select} = query
+    %{take: take, expr: expr} = select
+    {tag, from_take} = Map.get(take, 0, {:any, []})
+    source = get_source!(:select, query, 0)
+
+    # In from, if there is a schema and we have a map tag with preloads,
+    # it needs to be converted to a map in a later pass.
+    {take, from_tag} =
+      case tag do
+        :map when is_tuple(source) and elem(source, 1) != nil and preloads != [] ->
+          {Map.put(take, 0, {:struct, from_take}), :map}
+        _ ->
+          {take, :any}
+      end
 
-  defp normalize_fields(%{assocs: [], preloads: []} = query,
-                        %{take: take, expr: expr} = select) do
-    {fields, from} = collect_fields(expr, query, take, :error)
+    {postprocess, fields, from} =
+      collect_fields(expr, [], :error, query, take)
 
-    fields =
+    {fields, preprocess, postprocess} =
       case from do
-        {:ok, from} -> [select_source(0, from)|fields]
-        :error -> fields
+        {:ok, from_pre, from_taken} ->
+          {assoc_exprs, assoc_fields} = collect_assocs([], [], query, tag, from_take, assocs)
+          fields = from_taken ++ Enum.reverse(assoc_fields, Enum.reverse(fields))
+          preprocess = [from_pre | Enum.reverse(assoc_exprs)]
+          {fields, preprocess, {:from, from_tag, postprocess}}
+        :error when preloads != [] or assocs != [] ->
+          error! query, "the binding used in `from` must be selected in `select` when using `preload`"
+        :error ->
+          {Enum.reverse(fields), [], postprocess}
       end
 
-    %{select | fields: fields}
+    select = %{preprocess: preprocess, postprocess: postprocess, take: from_take, assocs: assocs}
+    {put_in(query.select.fields, fields), select}
   end
 
-  defp normalize_fields(%{assocs: assocs} = query,
-                        %{take: take, expr: expr} = select) do
-    {fields, from} = collect_fields(expr, query, take, :error)
-
-    case from do
-      {:ok, from} ->
-        {tag, take} = Map.get(take, 0, {:any, %{}})
-        assocs = collect_assocs(query, tag, take, assocs)
-        fields = [select_source(0, from)|assocs] ++ fields
-        %{select | fields: fields}
-      :error ->
-        error! query, "the binding used in `from` must be selected in `select` when using `preload`"
-    end
+  defp collect_fields({:&, _, [0]}, fields, :error, query, take) do
+    {expr, taken} = source_take!(:select, query, take, 0, 0)
+    {{:source, :from}, fields, {:ok, expr, taken}}
   end
-
-  defp collect_fields({:&, _, [0]}, query, take, :error) do
-    fields = take!(:select, query, take, 0, 0)
-    {[], {:ok, fields}}
+  defp collect_fields({:&, _, [0]}, fields, from, _query, _take) do
+    {{:source, :from}, fields, from}
   end
-  defp collect_fields({:&, _, [0]}, _query, _take, from) do
-    {[], from}
-  end
-  defp collect_fields({:&, _, [ix]}, query, take, from) do
-    fields = take!(:select, query, take, ix, ix)
-    {[select_source(ix, fields)], from}
+  defp collect_fields({:&, _, [ix]}, fields, from, query, take) do
+    {expr, taken} = source_take!(:select, query, take, ix, ix)
+    {expr, Enum.reverse(taken, fields), from}
   end
 
-  defp collect_fields({agg, meta, [{{:., _, [{:&, _, [ix]}, field]}, _, []} | _] = args},
-                      %{select: select} = query, _take, from) when agg in ~w(count avg min max sum)a do
+  defp collect_fields({agg, _, [{{:., _, [{:&, _, [ix]}, field]}, _, []} | _]} = expr,
+                      fields, from, %{select: select} = query, _take)
+       when agg in ~w(count avg min max sum)a do
     type =
+      # TODO: Support the :number type
       case agg do
         :count -> :integer
-        # TODO: Support the :number type
         :avg -> :any
         :sum -> :any
         _ -> source_type!(:select, query, select, ix, field)
       end
-    {[{agg, [ecto_type: type] ++ meta, args}], from}
+    {{:value, type}, [expr | fields], from}
   end
 
-  defp collect_fields({{:., _, [{:&, _, [ix]}, field]} = dot, meta, []},
-                      %{select: select} = query, _take, from) do
+  defp collect_fields({{:., _, [{:&, _, [ix]}, field]}, _, []} = expr,
+                      fields, from, %{select: select} = query, _take) do
     type = source_type!(:select, query, select, ix, field)
-    {[{dot, [ecto_type: type] ++ meta, []}], from}
-  end
-
-  defp collect_fields({left, right}, query, take, from) do
-    {left, from} = collect_fields(left, query, take, from)
-    {right, from} = collect_fields(right, query, take, from)
-    {left ++ right, from}
-  end
-  defp collect_fields({:{}, _, elems}, query, take, from),
-    do: collect_fields(elems, query, take, from)
-  defp collect_fields({:%{}, _, [{:|, _, [data, pairs]}]}, query, take, from),
-    do: collect_fields([data | pairs], query, take, from)
-  defp collect_fields({:%{}, _, pairs}, query, take, from),
-    do: collect_fields(pairs, query, take, from)
-  defp collect_fields({:%, _, [_name, expr]}, query, take, from),
-    do: collect_fields(expr, query, take, from)
-  defp collect_fields(list, query, take, from) when is_list(list),
-    do: Enum.flat_map_reduce(list, from, &collect_fields(&1, query, take, &2))
-  defp collect_fields(expr, _query, _take, from) when is_atom(expr) or is_binary(expr) or is_number(expr),
-    do: {[], from}
-  defp collect_fields(expr, _query, _take, from),
-    do: {[expr], from}
+    {{:value, type}, [expr | fields], from}
+  end
 
-  defp fetch_assoc(tag, take, assoc) do
-    case Access.fetch(take, assoc) do
-      {:ok, value} -> {:ok, {tag, value}}
-      :error -> :error
-    end
+  defp collect_fields({left, right}, fields, from, query, take) do
+    {args, fields, from} = collect_args([left, right], fields, from, query, take, [])
+    {{:tuple, args}, fields, from}
+  end
+
+  defp collect_fields({:{}, _, args}, fields, from, query, take) do
+    {args, fields, from} = collect_args(args, fields, from, query, take, [])
+    {{:tuple, args}, fields, from}
+  end
+
+  defp collect_fields({:%{}, _, [{:|, _, [data, args]}]}, fields, from, query, take) do
+    {data, fields, from} = collect_fields(data, fields, from, query, take)
+    {args, fields, from} = collect_kv(args, fields, from, query, take, [])
+    {{:map, data, args}, fields, from}
+  end
+
+  defp collect_fields({:%{}, _, args}, fields, from, query, take) do
+    {args, fields, from} = collect_kv(args, fields, from, query, take, [])
+    {{:map, args}, fields, from}
+  end
+
+  defp collect_fields({:%, _, [name, {:%{}, _, [{:|, _, [data, args]}]}]}, fields, from, query, take) do
+    {data, fields, from} = collect_fields(data, fields, from, query, take)
+    {args, fields, from} = collect_kv(args, fields, from, query, take, [])
+    struct!(name, args)
+    {{:struct, name, data, args}, fields, from}
+  end
+
+  defp collect_fields({:%, _, [name, {:%{}, _, args}]}, fields, from, query, take) do
+    {args, fields, from} = collect_kv(args, fields, from, query, take, [])
+    struct!(name, args)
+    {{:struct, name, args}, fields, from}
+  end
+
+  defp collect_fields(args, fields, from, query, take) when is_list(args) do
+    {args, fields, from} = collect_args(args, fields, from, query, take, [])
+    {{:list, args}, fields, from}
+  end
+
+  defp collect_fields(expr, fields, from, _query, _take)
+       when is_atom(expr) or is_binary(expr) or is_number(expr) do
+    {expr, fields, from}
+  end
+
+  defp collect_fields(%Ecto.Query.Tagged{tag: tag} = expr, fields, from, _query, _take) do
+    {{:value, tag}, [expr | fields], from}
+  end
+
+  defp collect_fields(expr, fields, from, _query, _take) do
+    {{:value, :any}, [expr | fields], from}
+  end
+
+  defp collect_kv([{key, value} | elems], fields, from, query, take, acc) do
+    {key, fields, from} = collect_fields(key, fields, from, query, take)
+    {value, fields, from} = collect_fields(value, fields, from, query, take)
+    collect_kv(elems, fields, from, query, take, [{key, value} | acc])
+  end
+  defp collect_kv([], fields, from, _query, _take, acc) do
+    {Enum.reverse(acc), fields, from}
+  end
+
+  defp collect_args([elem | elems], fields, from, query, take, acc) do
+    {elem, fields, from} = collect_fields(elem, fields, from, query, take)
+    collect_args(elems, fields, from, query, take, [elem | acc])
+  end
+  defp collect_args([], fields, from, _query, _take, acc) do
+    {Enum.reverse(acc), fields, from}
   end
 
-  defp collect_assocs(query, tag, take, [{assoc, {ix, children}}|tail]) do
+  defp collect_assocs(exprs, fields, query, tag, take, [{assoc, {ix, children}}|tail]) do
     case get_source!(:preload, query, ix) do
       {_, schema} = source when schema != nil ->
-        fields = take!(source, query, fetch_assoc(tag, take, assoc), assoc)
-        [select_source(ix, fields)] ++
-          collect_assocs(query, tag, fields, children) ++
-          collect_assocs(query, tag, take, tail)
+        {fetch, take_children} = fetch_assoc(tag, take, assoc)
+        {expr, taken} = take!(source, query, fetch, ix, assoc)
+        exprs = [expr | exprs]
+        fields = Enum.reverse(taken, fields)
+        {exprs, fields} = collect_assocs(exprs, fields, query, tag, take_children, children)
+        {exprs, fields} = collect_assocs(exprs, fields, query, tag, take, tail)
+        {exprs, fields}
       _ ->
         error! query, "can only preload sources with a schema " <>
                       "(fragments, binary and subqueries are not supported)"
     end
   end
-  defp collect_assocs(_query, _tag, _take, []) do
-    []
+  defp collect_assocs(exprs, fields, _query, _tag, _take, []) do
+    {exprs, fields}
   end
 
-  defp select_source(ix, nil), do: {:&, [], [ix, nil, nil]}
-  defp select_source(ix, fields) when is_list(fields) do
-    fields = for field <- fields, is_atom(field), do: field
-    {:&, [], [ix, fields, length(fields)]}
+  defp fetch_assoc(tag, take, assoc) do
+    case Access.fetch(take, assoc) do
+      {:ok, value} -> {{:ok, {tag, value}}, value}
+      :error -> {:error, []}
+    end
   end
 
-  defp take!(kind, query, take, field, ix) do
+  defp source_take!(kind, query, take, field, ix) do
     source = get_source!(kind, query, ix)
-    take!(source, query, Access.fetch(take, field), field)
+    take!(source, query, Access.fetch(take, field), ix, field)
   end
 
-  defp take!(source, query, fetched, field) do
+  defp take!(source, query, fetched, ix, field) do
     case {fetched, source} do
-      {{:ok, {:struct, _}}, {_, nil}} ->
-        error! query, "struct/2 in select expects a source with a schema"
       {{:ok, {_, []}}, {_, _}} ->
         error! query, "at least one field must be selected for binding `#{field}`, got an empty list"
-      {{:ok, {_, fields}}, {_, _}} ->
-        List.wrap(fields)
+
+      {{:ok, {:struct, _}}, {_, nil}} ->
+        error! query, "struct/2 in select expects a source with a schema"
+
+      {{:ok, {kind, fields}}, {source, schema}} ->
+        types = if schema, do: schema.__schema__(:types), else: %{}
+        types =
+          for field <- List.wrap(fields), is_atom(field) do
+            {field, Map.get(types, field, :any)}
+          end
+        schema = if kind == :map, do: nil, else: schema
+        {{:source, {source, schema}, types}, select_fields(types, ix)}
+
       {{:ok, {_, _}}, {:fragment, _, _}} ->
         error! query, "it is not possible to return a map/struct subset of a fragment, " <>
                       "you must explicitly return the desired individual fields"
+
       {{:ok, {_, _}}, %Ecto.SubQuery{}} ->
         error! query, "it is not possible to return a map/struct subset of a subquery, " <>
                       "you must explicitly select the whole subquery or individual fields only"
 
       {:error, {_, nil}} ->
-        nil # Checked by the adapter
-      {:error, {_, schema}} when schema != nil ->
-        schema.__schema__(:fields)
+        {{:value, :map}, [{:&, [], [ix]}]}
+
+      {:error, {_, schema}} ->
+        types = schema.__schema__(:types) |> Enum.to_list()
+        {{:source, source, types}, select_fields(types, ix)}
+
       {:error, {:fragment, _, _}} ->
-        nil # Checked by the adapter
-      {:error, %Ecto.SubQuery{fields: fields}} ->
-        Keyword.keys(fields)
+        {{:value, :map}, [{:&, [], [ix]}]}
+
+      {:error, %Ecto.SubQuery{select: select} = subquery} ->
+        {select, subquery |> subquery_types() |> select_fields(ix)}
+    end
+  end
+
+  defp select_fields(types, ix) do
+    for {field, _} <- types do
+      {{:., [], [{:&, [], [ix]}, field]}, [], []}
     end
   end
 
@@ -987,10 +1067,10 @@ defmodule Ecto.Query.Planner do
         source_type!(kind, query, expr, schema, field)
       {:fragment, _, _} ->
         :any
-      %Ecto.SubQuery{fields: fields, query: inner_query} ->
-        case Keyword.fetch(fields, field) do
-          {:ok, {{:., _, [{:&, _, [ix]}, field]}, _, []}} ->
-            source_type!(kind, inner_query, expr, ix, field)
+      %Ecto.SubQuery{} = subquery ->
+        case Keyword.fetch(subquery_types(subquery), field) do
+          {:ok, {:value, type}} ->
+            type
           {:ok, _} ->
             :any
           :error ->
diff --git a/lib/ecto/repo/preloader.ex b/lib/ecto/repo/preloader.ex
index 1941c0c6..439b1d08 100644
--- a/lib/ecto/repo/preloader.ex
+++ b/lib/ecto/repo/preloader.ex
@@ -94,7 +94,7 @@ defmodule Ecto.Repo.Preloader do
 
   defp maybe_pmap(assocs, repo, opts, fun) do
     if match?([_,_|_], assocs) and not repo.in_transaction? and
-       Keyword.get(opts, :in_parallel, true) do
+         Keyword.get(opts, :in_parallel, true) do
       # We pass caller: self() so pools like the ownership
       # pool knows where to fetch the connection from and
       # set the proper timeouts.
diff --git a/lib/ecto/repo/queryable.ex b/lib/ecto/repo/queryable.ex
index b336a452..91e47163 100644
--- a/lib/ecto/repo/queryable.ex
+++ b/lib/ecto/repo/queryable.ex
@@ -124,18 +124,19 @@ defmodule Ecto.Repo.Queryable do
     {meta, prepared, params} = Planner.query(query, operation, repo, adapter, 0)
 
     case meta do
-      %{fields: nil} ->
+      %{select: nil} ->
         adapter.execute(repo, meta, prepared, params, nil, opts)
-      %{select: select, fields: fields, prefix: prefix, take: take,
-        sources: sources, assocs: assocs, preloads: preloads} ->
-        preprocess    = preprocess(prefix, sources, adapter)
-        {count, rows} = adapter.execute(repo, meta, prepared, params, preprocess, opts)
-        postprocess   = postprocess(select, fields, take)
-        {_, take_0}   = Map.get(take, 0, {:any, %{}})
+      %{select: select, prefix: prefix, sources: sources, preloads: preloads} ->
+        %{preprocess: preprocess, postprocess: postprocess, take: take, assocs: assocs} = select
+        all_nil? = tuple_size(sources) != 1
+        preprocessor = &preprocess(&1, preprocess, all_nil?, prefix, adapter)
+        {count, rows} = adapter.execute(repo, meta, prepared, params, preprocessor, opts)
+        postprocessor = postprocessor(postprocess, take, prefix, adapter)
+
         {count,
           rows
           |> Ecto.Repo.Assoc.query(assocs, sources)
-          |> Ecto.Repo.Preloader.query(repo, preloads, take_0, postprocess, opts)}
+          |> Ecto.Repo.Preloader.query(repo, preloads, take, postprocessor, opts)}
     end
   end
 
@@ -143,177 +144,156 @@ defmodule Ecto.Repo.Queryable do
     {meta, prepared, params} = Planner.query(query, operation, repo, adapter, 0)
 
     case meta do
-      %{fields: nil} ->
-        adapter.stream(repo, meta, prepared, params, nil, opts)
-        |> Stream.flat_map(fn({_, nil}) -> [] end)
-      %{select: select, fields: fields, prefix: prefix, take: take,
-        sources: sources, assocs: assocs, preloads: preloads} ->
-        preprocess    = preprocess(prefix, sources, adapter)
-        stream        = adapter.stream(repo, meta, prepared, params, preprocess, opts)
-        postprocess   = postprocess(select, fields, take)
-        {_, take_0}   = Map.get(take, 0, {:any, %{}})
-
-        Stream.flat_map(stream, fn({_, rows}) ->
+      %{select: nil} ->
+        repo
+        |> adapter.stream(meta, prepared, params, nil, opts)
+        |> Stream.flat_map(fn {_, nil} -> [] end)
+      %{select: select, prefix: prefix, sources: sources, preloads: preloads} ->
+        %{preprocess: preprocess, postprocess: postprocess, take: take, assocs: assocs} = select
+        all_nil? = tuple_size(sources) != 1
+        preprocessor = &preprocess(&1, preprocess, all_nil?, prefix, adapter)
+        stream = adapter.stream(repo, meta, prepared, params, preprocessor, opts)
+        postprocessor = postprocessor(postprocess, take, prefix, adapter)
+
+        Stream.flat_map(stream, fn {_, rows} ->
           rows
           |> Ecto.Repo.Assoc.query(assocs, sources)
-          |> Ecto.Repo.Preloader.query(repo, preloads, take_0, postprocess, opts)
+          |> Ecto.Repo.Preloader.query(repo, preloads, take, postprocessor, opts)
         end)
     end
   end
 
-  defp preprocess(prefix, sources, adapter) do
-    &preprocess(&1, &2, prefix, &3, sources, adapter)
+  defp preprocess(row, [], _all_nil?, _prefix, _adapter) do
+    row
+  end
+  defp preprocess(row, [{:source, source_schema, fields} | sources], all_nil?, prefix, adapter) do
+    {entry, rest} = process_source(source_schema, fields, row, all_nil?, prefix, adapter)
+    [entry | preprocess(rest, sources, true, prefix, adapter)]
+  end
+  defp preprocess(row, [source | sources], all_nil?, prefix, adapter) do
+    {entry, rest} = process(row, source, nil, prefix, adapter)
+    [entry | preprocess(rest, sources, all_nil?, prefix, adapter)]
   end
 
-  defp preprocess({:&, _, [ix, fields, _]}, value, prefix, context, sources, adapter) do
-    case elem(sources, ix) do
-      {_source, nil} when is_map(value) ->
-        value
-      {_source, nil} when is_list(value) ->
-        load_schemaless(fields, value, %{})
-      {source, schema} when is_list(value) ->
-        Ecto.Schema.__load__(schema, prefix, source, context, {fields, value},
-                             &Ecto.Type.adapter_load(adapter, &1, &2))
-      {source, schema} when is_map(value) ->
-        Ecto.Schema.__load__(schema, prefix, source, context, value,
-                             &Ecto.Type.adapter_load(adapter, &1, &2))
-      %Ecto.SubQuery{meta: %{sources: sources, fields: fields, select: select, take: take}} ->
-        loaded = load_subquery(fields, value, prefix, context, sources, adapter)
-        postprocess(select, fields, take).(loaded)
+  defp postprocessor({:from, :any, postprocess}, _take, prefix, adapter) do
+    fn [from | row] ->
+      row |> process(postprocess, from, prefix, adapter) |> elem(0)
     end
   end
-
-  defp preprocess({agg, meta, [{{:., _, [{:&, _, [_]}, _]}, _, []}]},
-                  value, _prefix, _context, _sources, adapter) when agg in ~w(avg min max sum)a do
-    type = Keyword.fetch!(meta, :ecto_type)
-    load!(type, value, adapter)
+  defp postprocessor({:from, :map, postprocess}, take, prefix, adapter) do
+    fn [from | row] ->
+      row |> process(postprocess, to_map(from, take), prefix, adapter) |> elem(0)
+    end
   end
-
-  defp preprocess({{:., _, [{:&, _, [_]}, _]}, meta, []}, value, _prefix, _context, _sources, adapter) do
-    type = Keyword.fetch!(meta, :ecto_type)
-    load!(type, value, adapter)
+  defp postprocessor(postprocess, _take, prefix, adapter) do
+    fn row -> row |> process(postprocess, nil, prefix, adapter) |> elem(0) end
   end
 
-  defp preprocess(%Query.Tagged{tag: tag}, value, _prefix, _context, _sources, adapter) do
-    load!(tag, value, adapter)
+  defp process(row, {:struct, struct, data, args}, from, prefix, adapter) do
+    case process(row, data, from, prefix, adapter) do
+      {%{__struct__: ^struct} = data, row} ->
+        process_update(data, args, row, from, prefix, adapter)
+      {data, _row} ->
+        raise BadStructError, struct: struct, term: data
+    end
   end
-
-  defp preprocess(_key, value, _prefix, _context, _sources, _adapter) do
-    value
+  defp process(row, {:struct, struct, args}, from, prefix, adapter) do
+    {fields, row} = process_kv(args, row, from, prefix, adapter)
+    {Map.merge(struct.__struct__(), Map.new(fields)), row}
   end
-
-  defp load_subquery([{:&, [], [_, _, counter]} = field|fields], values, prefix, context, sources, adapter) do
-    {value, values} = Enum.split(values, counter)
-    [preprocess(field, value, prefix, context, sources, adapter) |
-     load_subquery(fields, values, prefix, context, sources, adapter)]
+  defp process(row, {:map, data, args}, from, prefix, adapter) do
+    {data, row} = process(row, data, from, prefix, adapter)
+    process_update(data, args, row, from, prefix, adapter)
   end
-  defp load_subquery([field|fields], [value|values], prefix, context, sources, adapter) do
-    [preprocess(field, value, prefix, context, sources, adapter) |
-     load_subquery(fields, values, prefix, context, sources, adapter)]
+  defp process(row, {:map, args}, from, prefix, adapter) do
+    {args, row} = process_kv(args, row, from, prefix, adapter)
+    {Map.new(args), row}
   end
-  defp load_subquery([], [], _prefix, _context, _sources, _adapter) do
-    []
+  defp process(row, {:list, args}, from, prefix, adapter) do
+    process_args(args, row, from, prefix, adapter)
   end
-
-  defp load_schemaless([field|fields], [value|values], acc),
-    do: load_schemaless(fields, values, Map.put(acc, field, value))
-  defp load_schemaless([], [], acc),
-    do: acc
-
-  defp load!(type, value, adapter) do
-    case Ecto.Type.adapter_load(adapter, type, value) do
-      {:ok, value} -> value
-      :error -> raise ArgumentError, "cannot load `#{inspect value}` as type #{inspect type}"
-    end
+  defp process(row, {:tuple, args}, from, prefix, adapter) do
+    {args, row} = process_args(args, row, from, prefix, adapter)
+    {List.to_tuple(args), row}
   end
-
-  defp postprocess(select, fields, take) do
-    # The planner always put the from as the first
-    # entry in the query, avoiding fetching it multiple
-    # times even if it appears multiple times in the query.
-    # So we always need to handle it specially.
-    from? = match?([{:&, _, [0, _, _]}|_], fields)
-    &postprocess(&1, select, take, from?)
+  defp process(row, {:source, :from}, from, _prefix, _adapter) do
+    {from, row}
   end
-
-  defp postprocess(row, expr, take, true),
-    do: transform_row(expr, take, hd(row), tl(row)) |> elem(0)
-  defp postprocess(row, expr, take, false),
-    do: transform_row(expr, take, nil, row) |> elem(0)
-
-  defp transform_row({:&, _, [0]}, take, from, values) do
-    {convert_to_tag(from, take[0]), values}
+  defp process(row, {:source, source_schema, fields}, _from, prefix, adapter) do
+    process_source(source_schema, fields, row, true, prefix, adapter)
   end
-
-  defp transform_row({:&, _, [ix]}, take, _from, [value|values]) do
-    {convert_to_tag(value, take[ix]), values}
+  defp process([value | row], {:value, :any}, _from, _prefix, _adapter) do
+    {value, row}
   end
-
-  defp transform_row({:{}, _, list}, take, from, values) do
-    {result, values} = transform_row(list, take, from, values)
-    {List.to_tuple(result), values}
+  defp process([value | row], {:value, type}, _from, _prefix, adapter) do
+    {load!(type, value, nil, nil, adapter), row}
   end
-
-  defp transform_row({left, right}, take, from, values) do
-    {[left, right], values} = transform_row([left, right], take, from, values)
-    {{left, right}, values}
+  defp process(row, value, _from, _prefix, _adapter)
+       when is_binary(value) or is_number(value) or is_atom(value) do
+    {value, row}
   end
 
-  defp transform_row({:%{}, _, [{:|, _, [data, pairs]}]}, take, from, values) do
-    {data, values} = transform_row(data, take, from, values)
-    Enum.reduce pairs, {data, values}, fn {k, v}, {data, acc} ->
-      {k, acc} = transform_row(k, take, from, acc)
-      {v, acc} = transform_row(v, take, from, acc)
-      {:maps.update(k, v, data), acc}
-    end
+  defp process_update(data, args, row, from, prefix, adapter) do
+    {args, row} = process_kv(args, row, from, prefix, adapter)
+    data = Enum.reduce(args, data, fn {key, value}, acc -> %{acc | key => value} end)
+    {data, row}
   end
 
-  defp transform_row({:%{}, _, pairs}, take, from, values) do
-    Enum.reduce pairs, {%{}, values}, fn {k, v}, {map, acc} ->
-      {k, acc} = transform_row(k, take, from, acc)
-      {v, acc} = transform_row(v, take, from, acc)
-      {Map.put(map, k, v), acc}
+  defp process_source({source, schema}, types, row, all_nil?, prefix, adapter) do
+    case split_values(types, row, [], all_nil?) do
+      {nil, row} ->
+        {nil, row}
+      {values, row} ->
+        struct = if schema, do: schema.__struct__(), else: %{}
+        loader = &Ecto.Type.adapter_load(adapter, &1, &2)
+        {Ecto.Schema.__safe_load__(struct, types, values, prefix, source, loader), row}
     end
   end
 
-  defp transform_row({:%, _, [name, map]}, take, from, values) do
-    case transform_row(map, take, from, values) do
-      {%{__struct__: ^name} = struct, acc} ->
-        {struct, acc}
-      {%{__struct__: _} = struct, _acc} ->
-        raise BadStructError, struct: name, term: struct
-      {map, acc} when is_map(map) ->
-        {struct(name, map), acc}
-    end
+  defp split_values([_ | types], [nil | values], acc, all_nil?) do
+    split_values(types, values, [nil | acc], all_nil?)
   end
-
-  defp transform_row(list, take, from, values) when is_list(list) do
-    Enum.map_reduce(list, values, &transform_row(&1, take, from, &2))
+  defp split_values([_ | types], [value | values], acc, _all_nil?) do
+    split_values(types, values, [value | acc], false)
   end
-
-  defp transform_row(expr, _take, _from, values)
-       when is_atom(expr) or is_binary(expr) or is_number(expr) do
-    {expr, values}
+  defp split_values([], values, _acc, true) do
+    {nil, values}
+  end
+  defp split_values([], values, acc, false) do
+    {Enum.reverse(acc), values}
   end
 
-  defp transform_row(_, _take, _from, [value|values]) do
-    {value, values}
+  defp process_args(args, row, from, prefix, adapter) do
+    Enum.map_reduce(args, row, fn arg, row ->
+      process(row, arg, from, prefix, adapter)
+    end)
   end
 
-  # We only need to worry about the struct -> map scenario.
-  # map -> struct is denied during compilation time.
-  # map -> map, struct -> struct and map/struct -> any are noop.
-  defp convert_to_tag(%{__struct__: _} = value, {:map, fields}),
-    do: to_map(value, fields)
-  defp convert_to_tag(value, _),
-    do: value
+  defp process_kv(kv, row, from, prefix, adapter) do
+    Enum.map_reduce(kv, row, fn {key, value}, row ->
+      {key, row} = process(row, key, from, prefix, adapter)
+      {value, row} = process(row, value, from, prefix, adapter)
+      {{key, value}, row}
+    end)
+  end
 
-  defp to_map(value, fields) when is_list(value) do
-    Enum.map(value, &to_map(&1, fields))
+  defp load!(type, value, field, struct, adapter) do
+    case Ecto.Type.adapter_load(adapter, type, value) do
+      {:ok, value} ->
+        value
+      :error ->
+        field = field && " for field #{inspect field}"
+        struct = struct && " in #{inspect struct}"
+        raise ArgumentError, "cannot load `#{inspect value}` as type #{inspect type}#{field}#{struct}"
+    end
   end
+
   defp to_map(nil, _fields) do
     nil
   end
+  defp to_map(value, fields) when is_list(value) do
+    Enum.map(value, &to_map(&1, fields))
+  end
   defp to_map(value, fields) do
     for field <- fields, into: %{} do
       case field do
diff --git a/lib/ecto/repo/schema.ex b/lib/ecto/repo/schema.ex
index fa359829..aecf6470 100644
--- a/lib/ecto/repo/schema.ex
+++ b/lib/ecto/repo/schema.ex
@@ -50,7 +50,7 @@ defmodule Ecto.Repo.Schema do
     {count, postprocess(rows, fields, adapter, schema, source)}
   end
 
-  defp preprocess([_|_] = fields, _schema),
+  defp preprocess([_ | _] = fields, _schema),
     do: fields
   defp preprocess([], _schema),
     do: raise ArgumentError, ":returning expects at least one field to be given, got an empty list"
@@ -63,13 +63,18 @@ defmodule Ecto.Repo.Schema do
 
   defp postprocess(nil, false, _adapter, _schema, _source), do: nil
   defp postprocess(rows, fields, _adapter, nil, _source) do
-    Enum.map(rows, &Map.new(Enum.zip(fields, &1)))
+    for row <- rows, do: Map.new(Enum.zip(fields, row))
   end
   defp postprocess(rows, fields, adapter, schema, {prefix, source}) do
-    Enum.map(rows, fn row ->
-      Ecto.Schema.__load__(schema, prefix, source, nil, {fields, row},
-                           &Ecto.Type.adapter_load(adapter, &1, &2))
-    end)
+    struct = schema.__struct__()
+
+    types = schema.__schema__(:types)
+    types = for field <- fields, do: {field, Map.fetch!(types, field)}
+
+    for row <- rows do
+      Ecto.Schema.__safe_load__(struct, types, row, prefix, source,
+                                &Ecto.Type.adapter_load(adapter, &1, &2))
+    end
   end
 
   defp extract_header_and_fields(rows, schema, autogenerate_id, adapter) do
@@ -376,9 +381,9 @@ defmodule Ecto.Repo.Schema do
   defp do_load(schema, {fields, values}, loader) when is_list(fields) and is_list(values),
     do: do_load(schema, Enum.zip(fields, values), loader)
   defp do_load(schema, data, loader) when is_atom(schema),
-    do: Ecto.Schema.__load__(schema, nil, nil, nil, data, loader)
+    do: Ecto.Schema.__unsafe_load__(schema, data, loader)
   defp do_load(types, data, loader) when is_map(types),
-    do: Ecto.Schema.__load__(%{}, types, data, loader)
+    do: Ecto.Schema.__unsafe_load__(%{}, types, data, loader)
 
   ## Helpers
 
@@ -459,8 +464,8 @@ defmodule Ecto.Repo.Schema do
                            "and #{inspect from} respectively"
     end
 
-    {Ecto.Query.Planner.normalize(query, :update_all, adapter, counter),
-     params, conflict_target}
+    {query, _} = Ecto.Query.Planner.normalize(query, :update_all, adapter, counter)
+    {query, params, conflict_target}
   end
 
   defp apply(%{valid?: false} = changeset, _adapter, _action, _args) do
diff --git a/lib/ecto/schema.ex b/lib/ecto/schema.ex
index b6b3b291..1ca7ae44 100644
--- a/lib/ecto/schema.ex
+++ b/lib/ecto/schema.ex
@@ -1406,46 +1406,52 @@ defmodule Ecto.Schema do
   defp type_to_module(other), do: other
 
   @doc false
-  def __load__(schema, prefix, source, context, data, loader) do
-    struct = schema.__struct__()
-    types = schema.__schema__(:types)
+  # Loads data into struct by assumes fields are properly
+  # named and belongs to the struct. Types and values are
+  # zipped together in one pass as they are loaded.
+  def __safe_load__(struct, types, values, prefix, source, loader) do
+    zipped = safe_load_zip(types, values, struct, loader)
 
-    case __load__(struct, types, data, loader) do
+    case Map.merge(struct, Map.new(zipped)) do
       %{__meta__: %Metadata{} = metadata} = struct ->
-        source = source || schema.__schema__(:source)
-        metadata = %{metadata | state: :loaded, source: {prefix, source}, context: context}
+        metadata = %{metadata | state: :loaded, source: {prefix, source}}
         Map.put(struct, :__meta__, metadata)
-      struct ->
-        struct
+      map ->
+        map
     end
   end
 
-  @doc false
-  def __load__(struct, types, map, loader) when is_map(map) do
-    Enum.reduce(types, struct, fn
-      {field, type}, acc ->
-        case fetch_string_or_atom_field(map, field) do
-          {:ok, value} -> Map.put(acc, field, load!(struct, field, type, value, loader))
-          :error -> acc
-        end
-    end)
+  defp safe_load_zip([{field, type} | fields], [value | values], struct, loader) do
+    [{field, load!(struct, field, type, value, loader)} |
+     safe_load_zip(fields, values, struct, loader)]
   end
-
-  def __load__(struct, types, {fields, values}, loader) when is_list(fields) and is_list(values) do
-    __load__(fields, values, struct, types, loader)
+  defp safe_load_zip([], [], _struct, _loader) do
+    []
   end
 
-  defp __load__([field|fields], [value|values], struct, types, loader) do
-    case Map.fetch(types, field) do
-      {:ok, type} ->
-        value = load!(struct, field, type, value, loader)
-        __load__(fields, values, Map.put(struct, field, value), types, loader)
-      :error ->
-        raise ArgumentError, "unknown field `#{field}`#{error_data(struct)}"
+  @doc false
+  # Assumes data does not all belongs to schema/struct
+  # and that it may also require source-based renaming.
+  def __unsafe_load__(schema, data, loader) do
+    types = schema.__schema__(:types)
+    struct = schema.__struct__()
+    case __unsafe_load__(struct, types, data, loader) do
+      %{__meta__: %Metadata{} = metadata} = struct ->
+        Map.put(struct, :__meta__, %{metadata | state: :loaded})
+      map ->
+        map
     end
   end
 
-  defp __load__([], [], struct, _types, _loader), do: struct
+  @doc false
+  def __unsafe_load__(struct, types, map, loader) when is_map(map) do
+    Enum.reduce(types, struct, fn {field, type}, acc ->
+      case fetch_string_or_atom_field(map, field) do
+        {:ok, value} -> Map.put(acc, field, load!(struct, field, type, value, loader))
+        :error -> acc
+      end
+    end)
+  end
 
   defp fetch_string_or_atom_field(map, field) when is_atom(field) do
     case Map.fetch(map, Atom.to_string(field)) do
diff --git a/lib/ecto/type.ex b/lib/ecto/type.ex
index 08c31d92..57ef358d 100644
--- a/lib/ecto/type.ex
+++ b/lib/ecto/type.ex
@@ -452,7 +452,7 @@ defmodule Ecto.Type do
   end
 
   defp load_embed(_field, schema, value, loader) when is_map(value) do
-    Ecto.Schema.__load__(schema, nil, nil, nil, value, loader)
+    Ecto.Schema.__unsafe_load__(schema, value, loader)
   end
 
   defp load_embed(field, _schema, value, _fun) do


Test patch:
diff --git a/test/ecto/adapters/mysql_test.exs b/test/ecto/adapters/mysql_test.exs
index 0a3e2232..34ee84fc 100644
--- a/test/ecto/adapters/mysql_test.exs
+++ b/test/ecto/adapters/mysql_test.exs
@@ -45,7 +45,8 @@ defmodule Ecto.Adapters.MySQLTest do
 
   defp normalize(query, operation \\ :all, counter \\ 0) do
     {query, _params, _key} = Ecto.Query.Planner.prepare(query, operation, Ecto.Adapters.MySQL, counter)
-    Ecto.Query.Planner.normalize(query, operation, Ecto.Adapters.MySQL, counter)
+    {query, _} = Ecto.Query.Planner.normalize(query, operation, Ecto.Adapters.MySQL, counter)
+    query
   end
 
   test "from" do
diff --git a/test/ecto/adapters/postgres_test.exs b/test/ecto/adapters/postgres_test.exs
index 3eaad565..6482cf3a 100644
--- a/test/ecto/adapters/postgres_test.exs
+++ b/test/ecto/adapters/postgres_test.exs
@@ -48,7 +48,8 @@ defmodule Ecto.Adapters.PostgresTest do
 
   defp normalize(query, operation \\ :all, counter \\ 0) do
     {query, _params, _key} = Ecto.Query.Planner.prepare(query, operation, Ecto.Adapters.Postgres, counter)
-    Ecto.Query.Planner.normalize(query, operation, Ecto.Adapters.Postgres, counter)
+    {query, _} = Ecto.Query.Planner.normalize(query, operation, Ecto.Adapters.Postgres, counter)
+    query
   end
 
   test "from" do
@@ -414,7 +415,7 @@ defmodule Ecto.Adapters.PostgresTest do
   test "update all with returning" do
     query = from(m in Schema, update: [set: [x: 0]]) |> select([m], m) |> normalize(:update_all)
     assert SQL.update_all(query) ==
-           ~s{UPDATE "schema" AS s0 SET "x" = 0 RETURNING s0."id", s0."x", s0."y", s0."z", s0."w"}
+           ~s{UPDATE "schema" AS s0 SET "x" = 0 RETURNING s0."id", s0."w", s0."x", s0."y", s0."z"}
   end
 
   test "update all array ops" do
@@ -456,7 +457,7 @@ defmodule Ecto.Adapters.PostgresTest do
 
   test "delete all with returning" do
     query = Schema |> Queryable.to_query |> select([m], m) |> normalize
-    assert SQL.delete_all(query) == ~s{DELETE FROM "schema" AS s0 RETURNING s0."id", s0."x", s0."y", s0."z", s0."w"}
+    assert SQL.delete_all(query) == ~s{DELETE FROM "schema" AS s0 RETURNING s0."id", s0."w", s0."x", s0."y", s0."z"}
   end
 
   test "delete all with prefix" do
@@ -565,6 +566,40 @@ defmodule Ecto.Adapters.PostgresTest do
            "SELECT s0.\"id\", s2.\"id\" FROM \"schema\" AS s0 INNER JOIN \"schema2\" AS s1 ON TRUE INNER JOIN \"schema2\" AS s2 ON TRUE"
   end
 
+  describe "query interpolation parameters" do
+    test "self join on subquery" do
+      subquery = select(Schema, [r], %{x: r.x, y: r.y})
+      query = subquery |> join(:inner, [c], p in subquery(subquery), true) |> normalize
+      assert SQL.all(query) ==
+             ~s{SELECT s0."x", s0."y" FROM "schema" AS s0 INNER JOIN } <>
+             ~s{(SELECT s0."x" AS "x", s0."y" AS "y" FROM "schema" AS s0) } <>
+             ~s{AS s1 ON TRUE}
+    end
+
+    test "self join on subquery with fragment" do
+      subquery = select(Schema, [r], %{string: fragment("downcase(?)", ^"string")})
+      query = subquery |> join(:inner, [c], p in subquery(subquery), true) |> normalize
+      assert SQL.all(query) ==
+             ~s{SELECT downcase($1) FROM "schema" AS s0 INNER JOIN } <>
+             ~s{(SELECT downcase($2) AS "string" FROM "schema" AS s0) } <>
+             ~s{AS s1 ON TRUE}
+    end
+
+    test "join on subquery with simple select" do
+      subquery = select(Schema, [r], %{x: ^999, w: ^888})
+      query = Schema
+              |> select([r], %{y: ^666})
+              |> join(:inner, [c], p in subquery(subquery), true)
+              |> where([a, b], a.x == ^111)
+              |> normalize
+
+      assert SQL.all(query) ==
+             ~s{SELECT $1 FROM "schema" AS s0 INNER JOIN } <>
+             ~s{(SELECT $2 AS "x", $3 AS "w" FROM "schema" AS s0) AS s1 ON TRUE } <>
+             ~s{WHERE (s0."x" = $4)}
+    end
+  end
+
   ## Associations
 
   test "association join belongs_to" do
diff --git a/test/ecto/query/inspect_test.exs b/test/ecto/query/inspect_test.exs
index c771b8e9..e1645d71 100644
--- a/test/ecto/query/inspect_test.exs
+++ b/test/ecto/query/inspect_test.exs
@@ -261,7 +261,8 @@ defmodule Ecto.Query.InspectTest do
 
   def plan(query) do
     {query, _params, _key} = Ecto.Query.Planner.prepare(query, :all, Ecto.TestAdapter, 0)
-    Ecto.Query.Planner.normalize(query, :all, Ecto.TestAdapter, 0)
+    {query, _} = Ecto.Query.Planner.normalize(query, :all, Ecto.TestAdapter, 0)
+    query
   end
 
   def i(query) do
diff --git a/test/ecto/query/planner_test.exs b/test/ecto/query/planner_test.exs
index 0f582452..8cd69824 100644
--- a/test/ecto/query/planner_test.exs
+++ b/test/ecto/query/planner_test.exs
@@ -47,9 +47,17 @@ defmodule Ecto.Query.PlannerTest do
 
   defp normalize_with_params(query, operation \\ :all) do
     {query, params, _key} = prepare(query, operation)
-    {query
-     |> Planner.returning(operation == :all)
-     |> Planner.normalize(operation, Ecto.TestAdapter, 0), params}
+    {query, _} =
+      query
+      |> Planner.returning(operation == :all)
+      |> Planner.normalize(operation, Ecto.TestAdapter, 0)
+    {query, params}
+  end
+
+  defp select_fields(fields, ix) do
+    for field <- fields do
+      {{:., [], [{:&, [], [ix]}, field]}, [], []}
+    end
   end
 
   test "prepare: merges all parameters" do
@@ -283,7 +291,7 @@ defmodule Ecto.Query.PlannerTest do
     {query, _, _} = prepare(from(p in subquery(Post), join: c in assoc(p, :comments)))
     assert [%{source: {"comments", Ecto.Query.PlannerTest.Comment}}] = query.joins
 
-    message = ~r/can only perform association joins on subqueries that return a single source in select/
+    message = ~r/can only perform association joins on subqueries that return a source with schema in select/
     assert_raise Ecto.QueryError, message, fn ->
       prepare(from(p in subquery(from p in Post, select: p.title), join: c in assoc(p, :comments)))
     end
@@ -313,32 +321,38 @@ defmodule Ecto.Query.PlannerTest do
   describe "prepare: subqueries select" do
     test "supports implicit select" do
       query = prepare(from(subquery(Post), [])) |> elem(0)
-      assert [{:id, {{:., [], [{:&, [], [0]}, :id]}, [], []}},
-              {:title, {{:., [], [{:&, [], [0]}, :title]}, [], []}} | _] = query.from.fields
+      assert "%Ecto.Query.PlannerTest.Post{id: &0.id(), title: &0.title(), " <>
+             "text: &0.text(), code: &0.code(), posted: &0.posted(), " <>
+             "visits: &0.visits(), links: &0.links()}" =
+             Macro.to_string(query.from.query.select.expr)
     end
 
     test "supports field selector" do
       query = from p in "posts", select: p.code
       query = prepare(from(subquery(query), [])) |> elem(0)
-      assert [code: {{:., [], [{:&, [], [0]}, :code]}, [], []}] = query.from.fields
+      assert "%{code: &0.code()}" =
+             Macro.to_string(query.from.query.select.expr)
 
       query = from p in Post, select: p.code
       query = prepare(from(subquery(query), [])) |> elem(0)
-      assert [code: {{:., [], [{:&, [], [0]}, :code]}, [], []}] = query.from.fields
+      assert "%{code: &0.code()}" =
+             Macro.to_string(query.from.query.select.expr)
     end
 
     test "supports maps" do
       query = from p in Post, select: %{code: p.code}
       query = prepare(from(subquery(query), [])) |> elem(0)
-      assert [code: {{:., [], [{:&, [], [0]}, :code]}, [], []}] = query.from.fields
+      assert "%{code: &0.code()}" =
+             Macro.to_string(query.from.query.select.expr)
     end
 
     test "supports update in maps" do
       query = from p in Post, select: %{p | code: p.title}
       query = prepare(from(subquery(query), [])) |> elem(0)
-      assert [{:id, {{:., [], [{:&, [], [0]}, :id]}, [], []}},
-              {:title, {{:., [], [{:&, [], [0]}, :title]}, [], []}} | _] = query.from.fields
-      assert {:code, {{:., [], [{:&, [], [0]}, :title]}, [], []}} = List.last(query.from.fields)
+      assert "%Ecto.Query.PlannerTest.Post{id: &0.id(), title: &0.title(), " <>
+             "text: &0.text(), posted: &0.posted(), visits: &0.visits(), " <>
+             "links: &0.links(), code: &0.title()}" =
+             Macro.to_string(query.from.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 ->
@@ -481,18 +495,20 @@ defmodule Ecto.Query.PlannerTest do
 
   test "normalize: select" do
     query = from(Post, []) |> normalize()
-    assert query.select.expr == {:&, [], [0]}
-    assert query.select.fields == [{:&, [], [0, [:id, :title, :text, :code, :posted, :visits, :links], 7]}]
+    assert query.select.expr ==
+             {:&, [], [0]}
+    assert query.select.fields ==
+           select_fields([:code, :id, :links, :posted, :text, :title, :visits], 0)
 
     query = from(Post, []) |> select([p], {p, p.title}) |> normalize()
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title, :text, :code, :posted, :visits, :links], 7]},
-            {{:., [], [{:&, [], [0]}, :title]}, [ecto_type: :string], []}]
+           select_fields([:code, :id, :links, :posted, :text, :title, :visits], 0) ++
+           [{{:., [], [{:&, [], [0]}, :title]}, [], []}]
 
     query = from(Post, []) |> select([p], {p.title, p}) |> normalize()
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title, :text, :code, :posted, :visits, :links], 7]},
-            {{:., [], [{:&, [], [0]}, :title]}, [ecto_type: :string], []}]
+           select_fields([:code, :id, :links, :posted, :text, :title, :visits], 0) ++
+           [{{:., [], [{:&, [], [0]}, :title]}, [], []}]
 
     query =
       from(Post, [])
@@ -501,9 +517,9 @@ defmodule Ecto.Query.PlannerTest do
       |> select([p, _], {p.title, p})
       |> normalize()
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title, :text, :code, :posted, :visits, :links], 7]},
-            {:&, [], [1, [:id, :text, :posted, :uuid, :post_id], 5]},
-            {{:., [], [{:&, [], [0]}, :title]}, [ecto_type: :string], []}]
+           select_fields([:code, :id, :links, :posted, :text, :title, :visits], 0) ++
+           select_fields([:id, :post_id, :posted, :text, :uuid], 1) ++
+           [{{:., [], [{:&, [], [0]}, :title]}, [], []}]
   end
 
   test "normalize: select with struct/2" do
@@ -513,12 +529,12 @@ defmodule Ecto.Query.PlannerTest do
 
     query = Post |> select([p], struct(p, [:id, :title])) |> normalize()
     assert query.select.expr == {:&, [], [0]}
-    assert query.select.fields == [{:&, [], [0, [:id, :title], 2]}]
+    assert query.select.fields == select_fields([:id, :title], 0)
 
     query = Post |> select([p], {struct(p, [:id, :title]), p.title}) |> normalize()
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title], 2]},
-            {{:., [], [{:&, [], [0]}, :title]}, [ecto_type: :string], []}]
+           select_fields([:id, :title], 0) ++
+           [{{:., [], [{:&, [], [0]}, :title]}, [], []}]
 
     query =
       Post
@@ -526,8 +542,8 @@ defmodule Ecto.Query.PlannerTest do
       |> select([p, c], {p, struct(c, [:id, :text])})
       |> normalize()
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title, :text, :code, :posted, :visits, :links], 7]},
-            {:&, [], [1, [:id, :text], 2]}]
+           select_fields([:code, :id, :links, :posted, :text, :title, :visits], 0) ++
+           select_fields([:id, :text], 1)
   end
 
   test "normalize: select with struct/2 on assoc" do
@@ -539,8 +555,8 @@ defmodule Ecto.Query.PlannerTest do
       |> normalize()
     assert query.select.expr == {:&, [], [0]}
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title], 2]},
-            {:&, [], [1, [:id, :text], 2]}]
+           select_fields([:id, :title], 0) ++
+           select_fields([:id, :text], 1)
 
     query =
       Post
@@ -550,21 +566,21 @@ defmodule Ecto.Query.PlannerTest do
       |> normalize()
     assert query.select.expr == {:&, [], [0]}
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title], 2]},
-            {:&, [], [1, [:id], 1]},
-            {:&, [], [1, [:id, :text], 2]},
-            {:&, [], [0, [:id], 1]}]
+           select_fields([:id, :title], 0) ++
+           select_fields([:id], 1) ++
+           select_fields([:id, :text], 1) ++
+           select_fields([:id], 0)
   end
 
   test "normalize: select with map/2" do
     query = Post |> select([p], map(p, [:id, :title])) |> normalize()
     assert query.select.expr == {:&, [], [0]}
-    assert query.select.fields == [{:&, [], [0, [:id, :title], 2]}]
+    assert query.select.fields == select_fields([:id, :title], 0)
 
     query = Post |> select([p], {map(p, [:id, :title]), p.title}) |> normalize()
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title], 2]},
-            {{:., [], [{:&, [], [0]}, :title]}, [ecto_type: :string], []}]
+           select_fields([:id, :title], 0) ++
+           [{{:., [], [{:&, [], [0]}, :title]}, [], []}]
 
     query =
       Post
@@ -572,8 +588,8 @@ defmodule Ecto.Query.PlannerTest do
       |> select([p, c], {p, map(c, [:id, :text])})
       |> normalize()
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title, :text, :code, :posted, :visits, :links], 7]},
-            {:&, [], [1, [:id, :text], 2]}]
+           select_fields([:code, :id, :links, :posted, :text, :title, :visits], 0) ++
+           select_fields([:id, :text], 1)
   end
 
   test "normalize: select with map/2 on assoc" do
@@ -585,8 +601,8 @@ defmodule Ecto.Query.PlannerTest do
       |> normalize()
     assert query.select.expr == {:&, [], [0]}
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title], 2]},
-            {:&, [], [1, [:id, :text], 2]}]
+           select_fields([:id, :title], 0) ++
+           select_fields([:id, :text], 1)
 
     query =
       Post
@@ -596,10 +612,10 @@ defmodule Ecto.Query.PlannerTest do
       |> normalize()
     assert query.select.expr == {:&, [], [0]}
     assert query.select.fields ==
-           [{:&, [], [0, [:id, :title], 2]},
-            {:&, [], [1, [:id], 1]},
-            {:&, [], [1, [:id, :text], 2]},
-            {:&, [], [0, [:id], 1]}]
+           select_fields([:id, :title], 0) ++
+           select_fields([:id], 1) ++
+           select_fields([:id, :text], 1) ++
+           select_fields([:id], 0)
   end
 
   test "normalize: preload" do
@@ -725,16 +741,16 @@ defmodule Ecto.Query.PlannerTest do
   test "normalize: merges subqueries fields when requested" do
     subquery = from p in Post, select: %{id: p.id, title: p.title}
     query = normalize(from(subquery(subquery), []))
-    assert query.select.fields == [{:&, [], [0, [:id, :title], 2]}]
+    assert query.select.fields == select_fields([:id, :title], 0)
 
     query = normalize(from(p in subquery(subquery), select: p.title))
-    assert query.select.fields == [{{:., [], [{:&, [], [0]}, :title]}, [ecto_type: :string], []}]
+    assert query.select.fields == [{{:., [], [{:&, [], [0]}, :title]}, [], []}]
 
     query = normalize(from(c in Comment, join: p in subquery(subquery), select: p))
-    assert query.select.fields == [{:&, [], [1, [:id, :title], 2]}]
+    assert query.select.fields == select_fields([:id, :title], 1)
 
     query = normalize(from(c in Comment, join: p in subquery(subquery), select: p.title))
-    assert query.select.fields == [{{:., [], [{:&, [], [1]}, :title]}, [ecto_type: :string], []}]
+    assert query.select.fields == [{{:., [], [{:&, [], [1]}, :title]}, [], []}]
 
     subquery = from p in Post, select: %{id: p.id, title: p.title}
     assert_raise Ecto.QueryError, ~r/it is not possible to return a map\/struct subset of a subquery/, fn ->