diff --git a/.formatter.exs b/.formatter.exs
index f722686a..996e2910 100644
--- a/.formatter.exs
+++ b/.formatter.exs
@@ -71,6 +71,7 @@ locals_without_parens = [
   list: 3,
   list: 4,
   load: 1,
+  manual: 1,
   manual?: 1,
   many_to_many: 2,
   many_to_many: 3,
diff --git a/lib/ash/actions/load.ex b/lib/ash/actions/load.ex
index 91dae928..ff876ba1 100644
--- a/lib/ash/actions/load.ex
+++ b/lib/ash/actions/load.ex
@@ -8,16 +8,18 @@ defmodule Ash.Actions.Load do
 
   def requests(
         query,
+        opts,
         root_query \\ nil,
         path \\ [],
         tenant \\ nil
       )
 
-  def requests(nil, _, _, _), do: {nil, []}
-  def requests(%{load: []} = query, _, _, _), do: {query, []}
+  def requests(nil, _opts, _, _, _), do: {nil, []}
+  def requests(%{load: []} = query, _opts, _, _, _), do: {query, []}
 
   def requests(
         %{load: loads} = query,
+        opts,
         root_query,
         path,
         tenant
@@ -61,6 +63,7 @@ defmodule Ash.Actions.Load do
       {related_query, further_requests} =
         requests(
           related_query,
+          opts,
           root_query,
           new_path,
           related_query.tenant
@@ -72,6 +75,7 @@ defmodule Ash.Actions.Load do
           further_requests ++
           do_requests(
             relationship,
+            opts,
             related_query,
             path,
             root_query
@@ -121,6 +125,14 @@ defmodule Ash.Actions.Load do
     data
   end
 
+  defp attach_to_many_loads(value, last_relationship, data, lead_path) when is_map(value) do
+    primary_key = Ash.Resource.Info.primary_key(last_relationship.source)
+
+    map_or_update(data, lead_path, fn record ->
+      Map.put(record, last_relationship.name, Map.get(value, Map.take(record, primary_key)))
+    end)
+  end
+
   defp attach_to_many_loads(value, last_relationship, data, lead_path) do
     values = Enum.group_by(value, &Map.get(&1, last_relationship.destination_field))
 
@@ -131,6 +143,14 @@ defmodule Ash.Actions.Load do
     end)
   end
 
+  defp attach_to_one_loads(value, last_relationship, data, lead_path) when is_map(value) do
+    primary_key = Ash.Resource.Info.primary_key(last_relationship.source)
+
+    map_or_update(data, lead_path, fn record ->
+      Map.put(record, last_relationship.name, Map.get(value, Map.take(record, primary_key)))
+    end)
+  end
+
   defp attach_to_one_loads(value, last_relationship, data, lead_path) do
     values =
       value
@@ -206,10 +226,11 @@ defmodule Ash.Actions.Load do
     last_relationship!(relationship.destination, rest)
   end
 
-  defp do_requests(relationship, related_query, path, root_query) do
+  defp do_requests(relationship, opts, related_query, path, root_query) do
     load_request =
       load_request(
         relationship,
+        opts,
         related_query,
         root_query,
         path
@@ -237,6 +258,7 @@ defmodule Ash.Actions.Load do
 
   defp load_request(
          relationship,
+         opts,
          related_query,
          root_query,
          path
@@ -302,65 +324,112 @@ defmodule Ash.Actions.Load do
           path,
           root_query
         ),
-      data:
-        Request.resolve(dependencies, fn data ->
-          base_query =
-            case get_in(data, request_path ++ [:authorization_filter]) do
-              nil ->
-                related_query
-
-              authorization_filter ->
-                Ash.Query.filter(related_query, ^authorization_filter)
-            end
-
-          source_query =
-            case path do
-              [] ->
-                root_query
-
-              path ->
-                get_in(data, [
-                  :load,
-                  Enum.reverse(Enum.map(path, &Map.get(&1, :name))),
-                  :query
-                ])
-            end
-
-          source_query =
-            if related_query.tenant do
-              Ash.Query.set_tenant(source_query, related_query.tenant)
-            else
-              source_query
-            end
-
-          with {:ok, new_query} <-
-                 true_load_query(
-                   relationship,
-                   base_query,
-                   data,
-                   path
-                 ),
-               {:ok, results} <-
-                 run_actual_query(
-                   new_query,
-                   base_query,
-                   data,
-                   path,
-                   relationship,
-                   source_query
-                 ) do
-            {:ok, results}
-          else
-            :nothing ->
-              {:ok, []}
-
-            {:error, error} ->
-              {:error, error}
-          end
-        end)
+      data: data(relationship, dependencies, request_path, related_query, path, root_query, opts)
     )
   end
 
+  defp data(
+         %{manual: manual} = relationship,
+         dependencies,
+         _request_path,
+         related_query,
+         path,
+         root_query,
+         request_opts
+       )
+       when not is_nil(manual) do
+    {mod, opts} =
+      case manual do
+        {mod, opts} ->
+          {mod, opts}
+
+        mod ->
+          {mod, []}
+      end
+
+    Request.resolve(dependencies, fn data ->
+      data =
+        case path do
+          [] ->
+            get_in(data, [:data, :data])
+
+          path ->
+            data
+            |> Map.get(:load, %{})
+            |> Map.get(Enum.reverse(Enum.map(Enum.drop(path, 1), & &1.name)), %{})
+            |> Map.get(:data, [])
+        end
+
+      mod.load(data, opts, %{
+        relationship: relationship,
+        query: related_query,
+        root_query: root_query,
+        actor: request_opts[:actor],
+        authorize: request_opts[:authorize?],
+        api: root_query.api,
+        tenant: related_query.tenant
+      })
+    end)
+  end
+
+  defp data(relationship, dependencies, request_path, related_query, path, root_query, _) do
+    Request.resolve(dependencies, fn data ->
+      base_query =
+        case get_in(data, request_path ++ [:authorization_filter]) do
+          nil ->
+            related_query
+
+          authorization_filter ->
+            Ash.Query.filter(related_query, ^authorization_filter)
+        end
+
+      source_query =
+        case path do
+          [] ->
+            root_query
+
+          path ->
+            get_in(data, [
+              :load,
+              Enum.reverse(Enum.map(path, &Map.get(&1, :name))),
+              :query
+            ])
+        end
+
+      source_query =
+        if related_query.tenant do
+          Ash.Query.set_tenant(source_query, related_query.tenant)
+        else
+          source_query
+        end
+
+      with {:ok, new_query} <-
+             true_load_query(
+               relationship,
+               base_query,
+               data,
+               path
+             ),
+           {:ok, results} <-
+             run_actual_query(
+               new_query,
+               base_query,
+               data,
+               path,
+               relationship,
+               source_query
+             ) do
+        {:ok, results}
+      else
+        :nothing ->
+          {:ok, []}
+
+        {:error, error} ->
+          {:error, error}
+      end
+    end)
+  end
+
   defp join_relationship(relationship) do
     Ash.Resource.Info.relationship(relationship.source, relationship.join_relationship)
   end
@@ -794,6 +863,11 @@ defmodule Ash.Actions.Load do
     end
   end
 
+  defp load_query(%{manual: manual}, related_query, _path, _root_query)
+       when not is_nil(manual) do
+    related_query
+  end
+
   defp load_query(
          relationship,
          related_query,
diff --git a/lib/ash/actions/read.ex b/lib/ash/actions/read.ex
index 4122fae6..7f335dac 100644
--- a/lib/ash/actions/read.ex
+++ b/lib/ash/actions/read.ex
@@ -285,7 +285,7 @@ defmodule Ash.Actions.Read do
               end
               |> run_before_action()
 
-            {query, load_requests} = Load.requests(query)
+            {query, load_requests} = Load.requests(query, opts)
 
             case Filter.run_other_data_layer_filters(
                    query.api,
diff --git a/lib/ash/actions/update.ex b/lib/ash/actions/update.ex
index 7d7dfedd..894e2938 100644
--- a/lib/ash/actions/update.ex
+++ b/lib/ash/actions/update.ex
@@ -197,13 +197,18 @@ defmodule Ash.Actions.Update do
                           {:ok, changeset.data, %{notifications: []}}
                         else
                           if Ash.Changeset.changing_attributes?(changeset) do
-                            changeset = Ash.Changeset.set_defaults(changeset, :update, true)
+                            changeset =
+                              changeset
+                              |> Ash.Changeset.set_defaults(:update, true)
+                              |> Ash.Changeset.put_context(:changed?, false)
 
                             resource
                             |> Ash.DataLayer.update(changeset)
                             |> add_tenant(changeset)
                             |> manage_relationships(api, changeset, engine_opts)
                           else
+                            changeset = Ash.Changeset.put_context(changeset, :changed?, false)
+
                             {:ok, changeset.data}
                             |> add_tenant(changeset)
                             |> manage_relationships(api, changeset, engine_opts)
diff --git a/lib/ash/changeset/changeset.ex b/lib/ash/changeset/changeset.ex
index 7b3b3a08..29548c33 100644
--- a/lib/ash/changeset/changeset.ex
+++ b/lib/ash/changeset/changeset.ex
@@ -1590,11 +1590,20 @@ defmodule Ash.Changeset do
 
         add_error(changeset, error)
 
+      %{manual: manual} = relationship when not is_nil(manual) ->
+        error =
+          InvalidRelationship.exception(
+            relationship: relationship.name,
+            message: "Cannot manage a manual relationship"
+          )
+
+        add_error(changeset, error)
+
       %{cardinality: :one, type: type} = relationship when is_list(input) and length(input) > 1 ->
         error =
           InvalidRelationship.exception(
             relationship: relationship.name,
-            message: "Cannot manage to a #{type} relationship with a list of records"
+            message: "Cannot manage a #{type} relationship with a list of records"
           )
 
         add_error(changeset, error)
diff --git a/lib/ash/query/query.ex b/lib/ash/query/query.ex
index 297a1d9e..0c41be0b 100644
--- a/lib/ash/query/query.ex
+++ b/lib/ash/query/query.ex
@@ -2022,6 +2022,22 @@ defmodule Ash.Query do
           )
         ]
 
+      %__MODULE__{} = destination_query ->
+        if Map.get(relationship, :manual) &&
+             (destination_query.limit ||
+                (destination_query.offset && destination_query.offset != 0)) do
+          [
+            InvalidQuery.exception(
+              resource: resource,
+              relationship: key,
+              query: destination_query,
+              load_path: Enum.reverse(path)
+            )
+          ]
+        else
+          do_validate_load(relationship.destination, destination_query, [key | path])
+        end
+
       other ->
         do_validate_load(relationship.destination, other, [key | path])
     end
diff --git a/lib/ash/resource/manual_relationship.ex b/lib/ash/resource/manual_relationship.ex
new file mode 100644
index 00000000..72892e2f
--- /dev/null
+++ b/lib/ash/resource/manual_relationship.ex
@@ -0,0 +1,28 @@
+defmodule Ash.Resource.ManualRelationship do
+  @moduledoc """
+  A module to implement manual relationships.
+  """
+
+  @type context :: %{
+          relationship: Ash.Resource.Relationships.relationship(),
+          query: Ash.Query.t(),
+          root_query: Ash.Query.t(),
+          actor: term,
+          tenant: term,
+          authorize?: term,
+          api: module
+        }
+
+  @callback load(
+              list(Ash.Resource.record()),
+              opts :: Keyword.t(),
+              context :: context()
+            ) ::
+              {:ok, map} | {:error, term}
+
+  defmacro __using__(_) do
+    quote do
+      @behaviour Ash.Resource.ManualRelationship
+    end
+  end
+end
diff --git a/lib/ash/resource/relationships/has_many.ex b/lib/ash/resource/relationships/has_many.ex
index a76c14c0..d91a7ff0 100644
--- a/lib/ash/resource/relationships/has_many.ex
+++ b/lib/ash/resource/relationships/has_many.ex
@@ -15,6 +15,7 @@ defmodule Ash.Resource.Relationships.HasMany do
     :read_action,
     :not_found_message,
     :violation_message,
+    :manual,
     validate_destination_field?: true,
     cardinality: :many,
     type: :has_many
@@ -33,7 +34,8 @@ defmodule Ash.Resource.Relationships.HasMany do
           destination_field: atom,
           private?: boolean,
           source_field: atom,
-          description: String.t()
+          description: String.t(),
+          manual: atom | {atom, Keyword.t()} | nil
         }
 
   import Ash.Resource.Relationships.SharedOptions
@@ -43,11 +45,18 @@ defmodule Ash.Resource.Relationships.HasMany do
                |> OptionsHelpers.set_default!(:source_field, :id)
 
   @opt_schema Ash.OptionsHelpers.merge_schemas(
-                [],
+                [
+                  manual()
+                ],
                 @global_opts,
                 "Relationship Options"
               )
 
   @doc false
   def opt_schema, do: @opt_schema
+
+  def manual({module, opts}) when is_atom(module) and is_list(opts),
+    do: {:ok, {module, opts}}
+
+  def manual(module) when is_atom(module), do: {:ok, {module, []}}
 end
diff --git a/lib/ash/resource/relationships/has_one.ex b/lib/ash/resource/relationships/has_one.ex
index 2723429f..62e31f94 100644
--- a/lib/ash/resource/relationships/has_one.ex
+++ b/lib/ash/resource/relationships/has_one.ex
@@ -17,6 +17,7 @@ defmodule Ash.Resource.Relationships.HasOne do
     :read_action,
     :not_found_message,
     :violation_message,
+    :manual,
     validate_destination_field?: true,
     cardinality: :one,
     type: :has_one,
@@ -37,7 +38,8 @@ defmodule Ash.Resource.Relationships.HasOne do
           private?: boolean,
           source_field: atom,
           allow_orphans?: boolean,
-          description: String.t()
+          description: String.t(),
+          manual: atom | {atom, Keyword.t()} | nil
         }
 
   import Ash.Resource.Relationships.SharedOptions
@@ -47,16 +49,17 @@ defmodule Ash.Resource.Relationships.HasOne do
                |> OptionsHelpers.set_default!(:source_field, :id)
 
   @opt_schema Ash.OptionsHelpers.merge_schemas(
-                [
-                  required?: [
-                    type: :boolean,
-                    doc: """
-                    Marks the relationship as required. This is *not* currently validated anywhere, since the
-                    relationship is managed by the destination, but ash_graphql uses it for type information,
-                    and it can be used for expressiveness.
-                    """
-                  ]
-                ],
+                [manual()] ++
+                  [
+                    required?: [
+                      type: :boolean,
+                      doc: """
+                      Marks the relationship as required. This is *not* currently validated anywhere, since the
+                      relationship is managed by the destination, but ash_graphql uses it for type information,
+                      and it can be used for expressiveness.
+                      """
+                    ]
+                  ],
                 @global_opts,
                 "Relationship Options"
               )
diff --git a/lib/ash/resource/relationships/shared_options.ex b/lib/ash/resource/relationships/shared_options.ex
index 761cad25..1feb7e75 100644
--- a/lib/ash/resource/relationships/shared_options.ex
+++ b/lib/ash/resource/relationships/shared_options.ex
@@ -92,4 +92,53 @@ defmodule Ash.Resource.Relationships.SharedOptions do
   def shared_options do
     @shared_options
   end
+
+  def manual do
+    {:manual,
+     type: {:ash_behaviour, Ash.Resource.ManualRelationship},
+     doc: """
+     Allows for relationships that are fetched manually. WARNING: EXPERIMENTAL
+
+     Manual relationships do not support filters or aggregates at the moment. In the future, what we may do is
+     allow the data layer to be configured with a hook that expresses how to implement this manual relationship
+     at the data layer level, like providing a custom ecto join for ash_postgres. This is the simple groundwork
+     for that.
+
+     ```elixir
+     # in the resource
+     relationships do
+       has_many :somethings, MyApp.Something do
+         manual {MyApp.FetchSomethings, [opt1: :value]}
+         # or if there are no opts
+         # manual MyApp.FetchSomethings
+       end
+     end
+
+     # the implementation
+     defmodule MyApp.FetchSomethings do
+       use Ash.Resource.ManualRelationship
+
+       def load(records, _opts, %{relationship: relationship}) do
+         # Return a map of primary keys of the records to the related records.
+         # This example is likely suboptimal because it does a separate fetch for
+         # each record, whereas you likely want to try to fetch them all at once,
+         # and then create the mapping from pkey values to related records
+         # For example:
+
+         # get the primary key
+         primary_key = Ash.Resource.Info.primary_key(relationship.source)
+         # e.g [:id]
+
+         # key the records by primary key and the related records with that primary key
+         {:ok,
+           Map.new(records, fn record ->
+             # the key is the pkey values, e.g `%{id: 1}`
+             # the value is the related records for that record
+             {Map.take(record, primary_key), get_related_records(record)}
+           end)}
+       end
+     end
+     ```
+     """}
+  end
 end
diff --git a/lib/ash/resource/transformers/validate_relationship_attributes.ex b/lib/ash/resource/transformers/validate_relationship_attributes.ex
index 4be5f30c..6b1a9455 100644
--- a/lib/ash/resource/transformers/validate_relationship_attributes.ex
+++ b/lib/ash/resource/transformers/validate_relationship_attributes.ex
@@ -16,6 +16,7 @@ defmodule Ash.Resource.Transformers.ValidateRelationshipAttributes do
 
     resource
     |> Ash.Resource.Info.relationships()
+    |> Enum.reject(&Map.get(&1, :manual))
     |> Enum.filter(& &1.validate_destination_field?)
     |> Enum.each(&validate_relationship(&1, attribute_names))
 
diff --git a/lib/ash/resource_formatter.ex b/lib/ash/resource_formatter.ex
index 21d78df0..b8db732c 100644
--- a/lib/ash/resource_formatter.ex
+++ b/lib/ash/resource_formatter.ex
@@ -101,19 +101,10 @@ defmodule Ash.ResourceFormatter do
               patches =
                 Enum.find_value(section_moves, fn {body_section, replacement_section} ->
                   if body_section != replacement_section do
-                    move_to =
-                      Enum.find_value(section_moves, fn {left, _} ->
-                        left == replacement_section && left
-                      end)
-
                     [
                       %{
                         range: Sourceror.get_range(body_section, include_comments: true),
-                        change: Sourceror.to_string(replacement_section, opts)
-                      },
-                      %{
-                        range: Sourceror.get_range(move_to, include_comments: true),
-                        change: Sourceror.to_string(body_section)
+                        change: Sourceror.to_string([replacement_section, body_section], opts)
                       }
                     ]
                   end
diff --git a/test/actions/load_test.exs b/test/actions/load_test.exs
index fe13f676..ee03cb78 100644
--- a/test/actions/load_test.exs
+++ b/test/actions/load_test.exs
@@ -36,6 +36,30 @@ defmodule Ash.Test.Actions.LoadTest do
     end
   end
 
+  defmodule PostsInSameCategory do
+    use Ash.Resource.ManualRelationship
+
+    def load(posts, _, %{query: destination_query, api: api}) do
+      categories = Enum.map(posts, & &1.category)
+
+      other_posts =
+        destination_query
+        |> Ash.Query.filter(category in ^categories)
+        |> api.read!()
+        |> Enum.group_by(& &1.category)
+
+      {:ok,
+       Map.new(posts, fn post ->
+         related_posts =
+           other_posts
+           |> Map.get(post.category, [])
+           |> Enum.reject(&(&1.id == post.id))
+
+         {Map.take(post, [:id]), related_posts}
+       end)}
+    end
+  end
+
   defmodule Post do
     @moduledoc false
     use Ash.Resource, data_layer: Ash.DataLayer.Ets
@@ -53,12 +77,17 @@ defmodule Ash.Test.Actions.LoadTest do
       uuid_primary_key :id
       attribute :title, :string
       attribute :contents, :string
+      attribute :category, :string
       timestamps()
     end
 
     relationships do
       belongs_to :author, Author
 
+      has_many :posts_in_same_category, __MODULE__ do
+        manual PostsInSameCategory
+      end
+
       many_to_many :categories, Ash.Test.Actions.LoadTest.Category,
         through: Ash.Test.Actions.LoadTest.PostCategory,
         destination_field_on_join_table: :category_id,
@@ -147,6 +176,29 @@ defmodule Ash.Test.Actions.LoadTest do
   end
 
   describe "loads" do
+    test "it allows loading manual relationships" do
+      post1 =
+        Post
+        |> new(%{title: "post1", category: "foo"})
+        |> Api.create!()
+
+      Post
+      |> new(%{title: "post2", category: "bar"})
+      |> Api.create!()
+
+      post3 =
+        Post
+        |> new(%{title: "post2", category: "foo"})
+        |> Api.create!()
+
+      post3_id = post3.id
+
+      assert [%{id: ^post3_id}] =
+               post1
+               |> Api.load!(:posts_in_same_category)
+               |> Map.get(:posts_in_same_category)
+    end
+
     test "it allows loading related data" do
       author =
         Author