diff --git a/lib/aggregate.ex b/lib/aggregate.ex
index af2bc82..57a4dcf 100644
--- a/lib/aggregate.ex
+++ b/lib/aggregate.ex
@@ -20,6 +20,8 @@ defmodule AshSql.Aggregate do
   def add_aggregates(query, aggregates, resource, select?, source_binding, root_data) do
     with {:ok, query, aggregates} <-
            AshSql.Aggregate.Common.normalize(query, aggregates, resource, root_data) do
+      query = track_projection_sources(query, aggregates, resource, source_binding)
+
       strategy(query, resource).add_aggregates(
         query,
         aggregates,
@@ -31,6 +33,24 @@ defmodule AshSql.Aggregate do
     end
   end
 
+  defp track_projection_sources(query, aggregates, resource, binding) do
+    if get_in(query.__ash_bindings__, [:bindings, binding, :projection]) do
+      Enum.reduce(aggregates, query, fn
+        %{related?: false}, query ->
+          query
+
+        %{relationship_path: [first | _]}, query ->
+          relationship = Ash.Resource.Info.relationship(resource, first)
+          AshSql.Query.Projection.track_source(query, binding, relationship)
+
+        _, query ->
+          query
+      end)
+    else
+      query
+    end
+  end
+
   def extract_shared_filters(aggregates) do
     AshSql.Aggregate.Lateral.extract_shared_filters(aggregates)
   end
diff --git a/lib/aggregate/grouped/query.ex b/lib/aggregate/grouped/query.ex
index 90fab94..c3dbbdc 100644
--- a/lib/aggregate/grouped/query.ex
+++ b/lib/aggregate/grouped/query.ex
@@ -62,7 +62,11 @@ defmodule AshSql.Aggregate.Grouped.Query do
 
       repo = AshSql.dynamic_repo(resource, implementation, query)
 
-      value = repo.one(query, AshSql.repo_opts(repo, implementation, nil, nil, resource))
+      value =
+        repo.one(
+          AshSql.Query.Projection.prune(query),
+          AshSql.repo_opts(repo, implementation, nil, nil, resource)
+        )
 
       {:ok, maybe_default_value(value, aggregate)}
     end
@@ -78,6 +82,7 @@ defmodule AshSql.Aggregate.Grouped.Query do
 
       result =
         query
+        |> AshSql.Query.Projection.prune()
         |> repo.one(AshSql.repo_opts(repo, implementation, nil, nil, resource))
         |> Map.get(aggregate.name)
 
diff --git a/lib/aggregate/lateral/query.ex b/lib/aggregate/lateral/query.ex
index bdc067d..89b1500 100644
--- a/lib/aggregate/lateral/query.ex
+++ b/lib/aggregate/lateral/query.ex
@@ -101,7 +101,11 @@ defmodule AshSql.Aggregate.Lateral.Query do
 
               _ ->
                 repo = AshSql.dynamic_repo(resource, implementation, query)
-                repo.one(group_query, AshSql.repo_opts(repo, implementation, nil, nil, resource))
+
+                repo.one(
+                  AshSql.Query.Projection.prune(group_query),
+                  AshSql.repo_opts(repo, implementation, nil, nil, resource)
+                )
             end
 
           {:ok, add_single_aggs(result, resource, query, cant_group, implementation)}
@@ -244,7 +248,7 @@ defmodule AshSql.Aggregate.Lateral.Query do
         Map.merge(
           result || %{},
           repo.one(
-            query,
+            AshSql.Query.Projection.prune(query),
             AshSql.repo_opts(repo, query.__ash_bindings__.sql_behaviour, nil, nil, resource)
           )
         )
diff --git a/lib/bindings.ex b/lib/bindings.ex
index e3dc417..82ae191 100644
--- a/lib/bindings.ex
+++ b/lib/bindings.ex
@@ -73,7 +73,14 @@ defmodule AshSql.Bindings do
       calculation_names: %{},
       context: context,
       root_binding: start_bindings,
-      bindings: %{start_bindings => %{path: [], type: :root, source: resource}}
+      bindings: %{
+        start_bindings => %{
+          path: [],
+          type: :root,
+          source: resource,
+          projection: context[:data_layer][:projection]
+        }
+      }
     })
   end
 
diff --git a/lib/expr.ex b/lib/expr.ex
index 52e4fa5..211b706 100644
--- a/lib/expr.ex
+++ b/lib/expr.ex
@@ -49,11 +49,13 @@ defmodule AshSql.Expr do
     Type
   }
 
+  alias AshSql.Query.Projection
+
   require Ecto.Query
 
   defmodule ExprInfo do
     @moduledoc false
-    defstruct has_error?: false
+    defstruct has_error?: false, projections: %{}
 
     @type t :: %__MODULE__{}
   end
@@ -109,8 +111,15 @@ defmodule AshSql.Expr do
     get_in(query.__ash_bindings__, [:context, :private, :as_of]) || DateTime.utc_now()
   end
 
-  defp do_dynamic_expr(query, expr, bindings, embedded?, acc, type \\ nil) do
+  defp backend_expr(query, expr, bindings, embedded?, acc, type) do
     case bindings.sql_behaviour.expr(query, expr, bindings, embedded?, acc, type) do
+      {:ok, expr, acc} -> {:ok, expr, Projection.unknown(acc)}
+      other -> other
+    end
+  end
+
+  defp do_dynamic_expr(query, expr, bindings, embedded?, acc, type \\ nil) do
+    case backend_expr(query, expr, bindings, embedded?, acc, type) do
       {:ok, expr, acc} -> {expr, acc}
       {:error, error} -> {:error, error}
       :error -> default_dynamic_expr(query, expr, bindings, embedded?, acc, type)
@@ -118,7 +127,7 @@ defmodule AshSql.Expr do
   end
 
   defp default_dynamic_expr(_, {:embed, other}, _bindings, _true, acc, _type) do
-    {other, acc}
+    {other, Projection.unknown(acc)}
   end
 
   defp default_dynamic_expr(query, %Not{expression: expression}, bindings, embedded?, acc, _type) do
@@ -634,7 +643,7 @@ defmodule AshSql.Expr do
          type
        )
        when is_list(right) do
-    case bindings.sql_behaviour.expr(query, expr, bindings, true, acc, type) do
+    case backend_expr(query, expr, bindings, true, acc, type) do
       {:ok, expr, acc} ->
         {expr, acc}
 
@@ -689,7 +698,7 @@ defmodule AshSql.Expr do
          _
        )
        when is_list(right) do
-    case bindings.sql_behaviour.expr(query, expr, bindings, true, acc, type) do
+    case backend_expr(query, expr, bindings, true, acc, type) do
       {:ok, expr, acc} ->
         {expr, acc}
 
@@ -1381,29 +1390,18 @@ defmodule AshSql.Expr do
         ]
       end)
 
-    do_dynamic_expr(
-      query,
-      %Fragment{
-        embedded?: pred_embedded?,
-        arguments:
-          [
-            raw: "(CASE WHEN ",
-            casted_expr: condition,
-            raw: " THEN ",
-            casted_expr: when_true
-          ] ++
-            additional_case_fragments ++
-            [
-              raw: " ELSE ",
-              casted_expr: when_false,
-              raw: " END)"
-            ]
-      },
-      bindings,
-      embedded?,
-      acc,
-      type
-    )
+    arguments =
+      [raw: "(CASE WHEN ", casted_expr: condition, raw: " THEN ", casted_expr: when_true] ++
+        additional_case_fragments ++
+        [raw: " ELSE ", casted_expr: when_false, raw: " END)"]
+
+    fragment = %Fragment{embedded?: pred_embedded?, arguments: arguments}
+
+    case backend_expr(query, fragment, bindings, embedded?, acc, type) do
+      {:ok, expr, acc} -> {expr, acc}
+      {:error, error} -> {:error, error}
+      :error -> build_fragment(query, arguments, bindings, pred_embedded? || embedded?, acc)
+    end
   end
 
   defp default_dynamic_expr(
@@ -1716,62 +1714,13 @@ defmodule AshSql.Expr do
          acc,
          _type
        ) do
-    arguments =
-      case arguments do
-        [{:raw, raw} | rest] ->
-          [{:raw, raw} | rest]
-
-        arguments ->
-          [{:raw, ""} | arguments]
-      end
-
-    arguments =
-      case List.last(arguments) do
-        nil ->
-          arguments
-
-        {:raw, _} ->
-          arguments
-
-        _ ->
-          arguments ++ [{:raw, ""}]
-      end
-
-    {params, fragment_data, _, acc} =
-      Enum.reduce(arguments, {[], [], 0, acc}, fn
-        {:raw, str}, {params, fragment_data, count, acc} ->
-          {params, [{:raw, str} | fragment_data], count, acc}
-
-        {:casted_expr, dynamic}, {params, fragment_data, count, acc} ->
-          {item, params, count} =
-            {{:^, [], [count]}, [{dynamic, :any} | params], count + 1}
-
-          {params, [{:expr, item} | fragment_data], count, acc}
-
-        {:expr, expr}, {params, fragment_data, count, acc} ->
-          {dynamic, acc} =
-            do_dynamic_expr(
-              query,
-              expr,
-              set_location(bindings, :sub_expr),
-              pred_embedded? || embedded?,
-              acc
-            )
-
-          {item, params, count} =
-            {{:^, [], [count]}, [{dynamic, :any} | params], count + 1}
-
-          {params, [{:expr, item} | fragment_data], count, acc}
-      end)
-
-    {%Ecto.Query.DynamicExpr{
-       fun: fn _query ->
-         {{:fragment, [], Enum.reverse(fragment_data)}, Enum.reverse(params), [], %{}}
-       end,
-       binding: [],
-       file: __ENV__.file,
-       line: __ENV__.line
-     }, acc}
+    build_fragment(
+      query,
+      arguments,
+      bindings,
+      pred_embedded? || embedded?,
+      Projection.unknown(acc)
+    )
   end
 
   defp default_dynamic_expr(
@@ -2432,6 +2381,8 @@ defmodule AshSql.Expr do
         field_name
       end
 
+    acc = if value, do: acc, else: Projection.reference(acc, bindings, ref_binding, field_name)
+
     expr =
       if value do
         value
@@ -2917,6 +2868,7 @@ defmodule AshSql.Expr do
          _expr_type
        ) do
     ref_binding = bindings.root_binding
+    acc = Projection.reference(acc, bindings, ref_binding, name)
 
     dynamic =
       if bindings[:parent?] &&
@@ -2949,6 +2901,8 @@ defmodule AshSql.Expr do
       reference_error!(query, ref)
     end
 
+    acc = Projection.reference(acc, bindings, ref_binding, name)
+
     constraints =
       if attr_type do
         constraints
@@ -3009,6 +2963,8 @@ defmodule AshSql.Expr do
       reference_error!(query, ref)
     end
 
+    acc = Projection.reference(acc, bindings, ref_binding, name)
+
     expr =
       if query.__ash_bindings__[:parent?] &&
            ref_binding not in List.wrap(bindings[:lateral_join_bindings]) do
@@ -3089,7 +3045,7 @@ defmodule AshSql.Expr do
         if is_list(other) do
           list_expr(query, other, bindings, true, acc, type)
         else
-          case bindings.sql_behaviour.expr(query, other, bindings, true, acc, type) do
+          case backend_expr(query, other, bindings, true, acc, type) do
             {:ok, expr, acc} ->
               {expr, acc}
 
@@ -3144,7 +3100,7 @@ defmodule AshSql.Expr do
       if is_list(value) do
         list_expr(query, value, bindings, false, acc, type)
       else
-        case bindings.sql_behaviour.expr(query, value, bindings, false, acc, type) do
+        case backend_expr(query, value, bindings, false, acc, type) do
           {:ok, expr, acc} ->
             {expr, acc}
 
@@ -3398,9 +3354,91 @@ defmodule AshSql.Expr do
         end
       )
 
+    acc =
+      cond do
+        Map.get(first_relationship, :manual) ->
+          Projection.unknown(acc)
+
+        Map.get(first_relationship, :no_attributes?) ->
+          acc
+
+        true ->
+          source_ref =
+            ref_binding(
+              %Ref{
+                attribute:
+                  Ash.Resource.Info.attribute(resource, first_relationship.source_attribute),
+                relationship_path: at_path,
+                resource: resource
+              },
+              bindings
+            )
+
+          Projection.reference(acc, bindings, source_ref, first_relationship.source_attribute)
+      end
+
     {Ecto.Query.dynamic(exists(subquery)), acc}
   end
 
+  defp build_fragment(query, arguments, bindings, embedded?, acc) do
+    arguments =
+      case arguments do
+        [{:raw, raw} | rest] ->
+          [{:raw, raw} | rest]
+
+        arguments ->
+          [{:raw, ""} | arguments]
+      end
+
+    arguments =
+      case List.last(arguments) do
+        nil ->
+          arguments
+
+        {:raw, _} ->
+          arguments
+
+        _ ->
+          arguments ++ [{:raw, ""}]
+      end
+
+    {params, fragment_data, _, acc} =
+      Enum.reduce(arguments, {[], [], 0, acc}, fn
+        {:raw, str}, {params, fragment_data, count, acc} ->
+          {params, [{:raw, str} | fragment_data], count, acc}
+
+        {:casted_expr, dynamic}, {params, fragment_data, count, acc} ->
+          {item, params, count} =
+            {{:^, [], [count]}, [{dynamic, :any} | params], count + 1}
+
+          {params, [{:expr, item} | fragment_data], count, acc}
+
+        {:expr, expr}, {params, fragment_data, count, acc} ->
+          {dynamic, acc} =
+            do_dynamic_expr(
+              query,
+              expr,
+              set_location(bindings, :sub_expr),
+              embedded?,
+              acc
+            )
+
+          {item, params, count} =
+            {{:^, [], [count]}, [{dynamic, :any} | params], count + 1}
+
+          {params, [{:expr, item} | fragment_data], count, acc}
+      end)
+
+    {%Ecto.Query.DynamicExpr{
+       fun: fn _query ->
+         {{:fragment, [], Enum.reverse(fragment_data)}, Enum.reverse(params), [], %{}}
+       end,
+       binding: [],
+       file: __ENV__.file,
+       line: __ENV__.line
+     }, acc}
+  end
+
   defp range_operand(query, operand, nil, bindings, embedded?, acc) do
     do_dynamic_expr(query, operand, set_location(bindings, :sub_expr), embedded?, acc, nil)
   end
@@ -3741,7 +3779,7 @@ defmodule AshSql.Expr do
 
       case bindings.sql_behaviour.list_expr(query, value, bindings, embedded?, acc, type) do
         {:ok, expr, acc} ->
-          {expr, acc}
+          {expr, Projection.unknown(acc)}
 
         {:error, error} ->
           raise "Error while building list expression: #{error}"
@@ -4146,10 +4184,12 @@ defmodule AshSql.Expr do
   end
 
   @doc false
-  def merge_accumulator(%ExprInfo{has_error?: left_has_error?}, %ExprInfo{
-        has_error?: right_has_error?
-      }) do
-    %ExprInfo{has_error?: left_has_error? || right_has_error?}
+  def merge_accumulator(%ExprInfo{} = left, %ExprInfo{} = right) do
+    %ExprInfo{
+      has_error?: left.has_error? || right.has_error?,
+      projections:
+        Projection.merge(Map.get(left, :projections, %{}), Map.get(right, :projections, %{}))
+    }
   end
 
   @doc false
diff --git a/lib/join.ex b/lib/join.ex
index fbfbeda..b4fd452 100644
--- a/lib/join.ex
+++ b/lib/join.ex
@@ -694,6 +694,18 @@ defmodule AshSql.Join do
             )
         end
 
+      query =
+        if projection = opts[:projection] do
+          query
+          |> put_in([Access.key!(:__ash_bindings__), :projection], projection)
+          |> update_in(
+            [Access.key!(:__ash_bindings__), :bindings, query.__ash_bindings__.root_binding],
+            &Map.put(&1, :projection, projection)
+          )
+        else
+          query
+        end
+
       {:ok, query}
     end
   end
@@ -746,7 +758,8 @@ defmodule AshSql.Join do
     |> Ash.Query.set_context(%{
       data_layer: %{
         table: nil,
-        start_bindings_at: opts[:start_bindings_at] || 0
+        start_bindings_at: opts[:start_bindings_at] || 0,
+        projection: opts[:projection]
       }
     })
     |> then(fn query ->
@@ -1277,6 +1290,8 @@ defmodule AshSql.Join do
           end
         )
 
+      query = AshSql.Query.Projection.track_source(query, current_binding, relationship)
+
       case apply(module, query.__ash_bindings__.sql_behaviour.manual_relationship_function(), [
              query,
              opts,
@@ -1286,6 +1301,12 @@ defmodule AshSql.Join do
              relationship_destination
            ]) do
         {:ok, query} ->
+          query =
+            update_in(
+              query.__ash_bindings__.expression_accumulator,
+              &AshSql.Query.Projection.unknown/1
+            )
+
           AshSql.Aggregate.add_aggregates(
             query,
             used_aggregates,
@@ -1420,6 +1441,8 @@ defmodule AshSql.Join do
             end
           )
 
+        query = AshSql.Query.Projection.track_source(query, current_binding, relationship)
+
         query =
           case kind do
             :inner ->
@@ -1477,7 +1500,8 @@ defmodule AshSql.Join do
     full_path = path ++ [relationship.name]
     initial_ash_bindings = query.__ash_bindings__
 
-    binding_data = %{type: kind, path: full_path, source: source}
+    projection = if Map.get(relationship, :from_many?), do: make_ref()
+    binding_data = %{type: kind, path: full_path, source: source, projection: projection}
 
     query = AshSql.Bindings.add_binding(query, binding_data)
 
@@ -1506,6 +1530,8 @@ defmodule AshSql.Join do
         end
       )
 
+    query = AshSql.Query.Projection.track_source(query, current_binding, relationship)
+
     destination_filter =
       relationship.destination
       |> Ash.Query.do_filter(relationship.filter, parent_stack: [query.__ash_bindings__.resource])
@@ -1540,6 +1566,7 @@ defmodule AshSql.Join do
              refs_at_path: path,
              require_lateral?: require_lateral?,
              filter_subquery?: true,
+             projection: projection,
              join_binding:
                if(!Map.get(relationship, :from_many?), do: initial_ash_bindings.current),
              on_subquery: fn subquery ->
@@ -1720,6 +1747,8 @@ defmodule AshSql.Join do
         end
       )
 
+    query = AshSql.Query.Projection.track_source(query, current_binding, relationship)
+
     query_with_bindings =
       AshSql.Bindings.add_binding(query, %{
         path: join_path,
diff --git a/lib/query/projection.ex b/lib/query/projection.ex
new file mode 100644
index 0000000..ce94338
--- /dev/null
+++ b/lib/query/projection.ex
@@ -0,0 +1,161 @@
+# SPDX-FileCopyrightText: 2024 ash_sql contributors <https://github.com/ash-project/ash_sql/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshSql.Query.Projection do
+  @moduledoc false
+  import Ecto.Query, only: [from: 2]
+
+  @clauses [
+    :select,
+    :wheres,
+    :havings,
+    :group_bys,
+    :order_bys,
+    :distinct,
+    :windows,
+    :limit,
+    :offset
+  ]
+
+  def reference(acc, bindings, binding, field) do
+    with token when not is_nil(token) <- get_in(bindings, [:bindings, binding, :projection]),
+         refs when is_map(refs) <- Map.get(acc, :projections, %{}),
+         true <- is_atom(field) and not is_nil(field) do
+      Map.put(
+        acc,
+        :projections,
+        Map.update(refs, token, MapSet.new([field]), &MapSet.put(&1, field))
+      )
+    else
+      false -> unknown(acc)
+      _ -> acc
+    end
+  end
+
+  def track_source(query, _binding, %{no_attributes?: true}), do: query
+
+  def track_source(query, binding, %{source_attribute: field}) when not is_nil(field) do
+    if get_in(query.__ash_bindings__, [:bindings, binding, :projection]) do
+      update_in(query.__ash_bindings__.expression_accumulator, fn acc ->
+        reference(acc, query.__ash_bindings__, binding, field)
+      end)
+    else
+      query
+    end
+  end
+
+  def track_source(query, _binding, _relationship), do: query
+
+  # Unknown dependencies leave the read's original selection intact.
+  def unknown(acc), do: Map.put(acc, :projections, :unknown)
+  def merge(:unknown, _), do: :unknown
+  def merge(_, :unknown), do: :unknown
+
+  def merge(left, right),
+    do: Map.merge(left, right, fn _, left, right -> MapSet.union(left, right) end)
+
+  def prune(query) do
+    case collect(query, %{}) do
+      :unknown -> query
+      refs when map_size(refs) == 0 -> query
+      refs -> rewrite(query, refs)
+    end
+  end
+
+  # A read action may change the Ecto query after references have been recorded.
+  defp collect(%{__ash_bindings__: %{context: %{action: %{modify_query: modify}}}}, _refs)
+       when not is_nil(modify), do: :unknown
+
+  # Only visit query nodes and their metadata, never expressions or parameter values.
+  defp collect(%{__ash_bindings__: %{expression_accumulator: acc}} = query, refs) do
+    if query.assocs != [] || query.preloads != [] || query.combinations != [] || query.with_ctes do
+      :unknown
+    else
+      refs = merge(refs, Map.get(acc, :projections, %{}))
+
+      Enum.reduce_while(children(query), refs, fn child, refs ->
+        case collect(child, refs) do
+          :unknown -> {:halt, :unknown}
+          refs -> {:cont, refs}
+        end
+      end)
+    end
+  end
+
+  defp collect(_, _), do: :unknown
+
+  defp children(query) do
+    [query.from && query.from.source | Enum.flat_map(query.joins, &[&1.source, &1.on])]
+    |> Kernel.++(Enum.map(@clauses, &Map.get(query, &1)))
+    |> Enum.flat_map(&subqueries/1)
+  end
+
+  defp subqueries(%Ecto.SubQuery{query: query}), do: [query]
+  defp subqueries(%{from: _, joins: _} = query), do: [query]
+  defp subqueries(%{subqueries: subqueries}), do: Enum.flat_map(subqueries, &subqueries/1)
+  defp subqueries({_name, value}), do: subqueries(value)
+  defp subqueries(list) when is_list(list), do: Enum.flat_map(list, &subqueries/1)
+  defp subqueries(_), do: []
+
+  defp rewrite(query, refs) do
+    mapper = &map_subqueries(&1, refs)
+    from = if query.from, do: %{query.from | source: mapper.(query.from.source)}, else: nil
+    joins = Enum.map(query.joins, &%{&1 | source: mapper.(&1.source), on: mapper.(&1.on)})
+    query = %{query | from: from, joins: joins}
+    query = Enum.reduce(@clauses, query, &Map.update!(&2, &1, mapper))
+
+    case get_in(query, [Access.key(:__ash_bindings__, %{}), :projection]) do
+      nil -> query
+      token -> restrict(query, Map.get(refs, token, MapSet.new()))
+    end
+  end
+
+  defp map_subqueries(%Ecto.SubQuery{query: query} = subquery, refs),
+    do: %{subquery | query: rewrite(query, refs)}
+
+  defp map_subqueries(%{from: _, joins: _} = query, refs), do: rewrite(query, refs)
+
+  defp map_subqueries(%{subqueries: subqueries} = clause, refs),
+    do: %{clause | subqueries: Enum.map(subqueries, &map_subqueries(&1, refs))}
+
+  defp map_subqueries({name, value}, refs), do: {name, map_subqueries(value, refs)}
+
+  defp map_subqueries(list, refs) when is_list(list),
+    do: Enum.map(list, &map_subqueries(&1, refs))
+
+  defp map_subqueries(value, _refs), do: value
+
+  defp restrict(
+         %{
+           select: nil,
+           distinct: nil,
+           group_bys: [],
+           havings: [],
+           combinations: [],
+           with_ctes: nil
+         } = query,
+         fields
+       ) do
+    case query.from.source do
+      %Ecto.SubQuery{query: inner} = subquery ->
+        %{query | from: %{query.from | source: %{subquery | query: restrict(inner, fields)}}}
+
+      {table, schema} when is_binary(table) and is_atom(schema) and not is_nil(schema) ->
+        attributes = schema.__schema__(:fields)
+        selected = Enum.filter(attributes, &MapSet.member?(fields, &1))
+
+        if selected != [] && length(selected) == MapSet.size(fields) &&
+             length(selected) < length(attributes) do
+          from(row in query, select: struct(row, ^selected))
+        else
+          query
+        end
+
+      _ ->
+        query
+    end
+  end
+
+  defp restrict(query, _fields), do: query
+end
diff --git a/test/projection_test.exs b/test/projection_test.exs
new file mode 100644
index 0000000..ca813fd
--- /dev/null
+++ b/test/projection_test.exs
@@ -0,0 +1,51 @@
+# SPDX-FileCopyrightText: 2024 ash_sql contributors <https://github.com/ash-project/ash_sql/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshSql.Query.ProjectionTest do
+  use ExUnit.Case, async: true
+  import Ecto.Query
+  alias AshSql.Query.Projection
+
+  defmodule Comment do
+    use Ecto.Schema
+
+    schema "comments" do
+      field(:post_id, :integer)
+      field(:title, :string)
+      field(:likes, :integer)
+    end
+  end
+
+  defp tracked_query(refs_fun \\ & &1) do
+    token = make_ref()
+    metadata = %{expression_accumulator: %{projections: %{}}}
+    inner = from(c in Comment, limit: 1) |> Map.put(:__ash_bindings__, metadata)
+
+    related =
+      from(c in subquery(inner))
+      |> Map.put(:__ash_bindings__, Map.put(metadata, :projection, token))
+
+    from(p in "posts", left_lateral_join: c in ^related, on: true, select: c.title)
+    |> Map.put(:__ash_bindings__, %{
+      expression_accumulator: %{projections: refs_fun.(%{token => MapSet.new([:title])})}
+    })
+  end
+
+  test "uses recorded fields without modifying the input query" do
+    query = tracked_query()
+    pruned = Projection.prune(query)
+    assert hd(pruned.joins).source.from.source.query.select.take == %{0 => {:struct, [:title]}}
+    assert hd(query.joins).source.from.source.query.select == nil
+  end
+
+  test "unknown references leave the entire query unchanged" do
+    query = tracked_query(fn _ -> :unknown end)
+    assert Projection.prune(query) == query
+  end
+
+  test "queries without tracking metadata pass through" do
+    query = tracked_query() |> Map.delete(:__ash_bindings__)
+    assert Projection.prune(query) == query
+  end
+end