diff --git a/lib/ash/actions/aggregate.ex b/lib/ash/actions/aggregate.ex
index a5939cda3..2a9d1a7f2 100644
--- a/lib/ash/actions/aggregate.ex
+++ b/lib/ash/actions/aggregate.ex
@@ -174,17 +174,29 @@ defmodule Ash.Actions.Aggregate do
     end)
     |> case do
       {:ok, aggregates} ->
-        {:ok,
-         Enum.map(aggregates, fn aggregate ->
-           Ash.Actions.Read.add_calc_context(
-             aggregate,
-             opts[:actor],
-             opts[:authorize?],
-             opts[:tenant],
-             opts[:tracer],
-             query.domain
-           )
-         end)}
+        Enum.reduce_while(aggregates, {:ok, []}, fn aggregate, {:ok, aggregates} ->
+          if Ash.DataLayer.data_layer_can?(aggregate.resource, {:query_aggregate, aggregate.kind}) do
+            aggregate =
+              Ash.Actions.Read.add_calc_context(
+                aggregate,
+                opts[:actor],
+                opts[:authorize?],
+                opts[:tenant],
+                opts[:tracer],
+                query.domain
+              )
+
+            {:cont, {:ok, [aggregate | aggregates]}}
+          else
+            {:halt,
+             {:error,
+              Ash.Error.Query.AggregatesNotSupported.exception(
+                resource: aggregate.resource,
+                feature: "using",
+                type: :query_aggregate
+              )}}
+          end
+        end)
 
       other ->
         other
diff --git a/lib/ash/error/query/aggregates_not_supported.ex b/lib/ash/error/query/aggregates_not_supported.ex
index 546824d51..3598e38f4 100644
--- a/lib/ash/error/query/aggregates_not_supported.ex
+++ b/lib/ash/error/query/aggregates_not_supported.ex
@@ -2,9 +2,18 @@ defmodule Ash.Error.Query.AggregatesNotSupported do
   @moduledoc "Used when the data_layer does not support aggregates, or filtering/sorting them"
   use Ash.Error.Exception
 
-  use Splode.Error, fields: [:resource, :feature], class: :invalid
+  use Splode.Error, fields: [:resource, :feature, type: :aggregate], class: :invalid
 
-  def message(%{resource: resource, feature: feature}) do
-    "Data layer for #{inspect(resource)} does not support #{feature} aggregates"
+  def message(%{resource: resource, feature: feature, type: type}) do
+    type =
+      case type do
+        :aggregate ->
+          "aggregates"
+
+        :query_aggregate ->
+          "query aggregates"
+      end
+
+    "Data layer for #{inspect(resource)} does not support #{feature} #{type}"
   end
 end
diff --git a/lib/ash/filter/filter.ex b/lib/ash/filter/filter.ex
index b1dc43aa4..77b19afd4 100644
--- a/lib/ash/filter/filter.ex
+++ b/lib/ash/filter/filter.ex
@@ -2783,56 +2783,65 @@ defmodule Ash.Filter do
         end
 
       aggregate = aggregate(context, field) ->
-        related = Ash.Resource.Info.related(context.resource, aggregate.relationship_path)
+        if Ash.DataLayer.data_layer_can?(context.resource, {:aggregate, aggregate.kind}) do
+          related = Ash.Resource.Info.related(context.resource, aggregate.relationship_path)
+
+          read_action =
+            aggregate.read_action || Ash.Resource.Info.primary_action!(related, :read).name
+
+          with %{valid?: true} = aggregate_query <- Ash.Query.for_read(related, read_action),
+               %{valid?: true} = aggregate_query <-
+                 Ash.Query.Aggregate.build_query(
+                   aggregate_query,
+                   context.resource,
+                   filter: aggregate.filter,
+                   sort: aggregate.sort
+                 ),
+               {:ok, query_aggregate} <-
+                 Aggregate.new(
+                   context.resource,
+                   aggregate.name,
+                   aggregate.kind,
+                   agg_name: aggregate.name,
+                   path: aggregate.relationship_path,
+                   query: aggregate_query,
+                   field: aggregate.field,
+                   default: aggregate.default,
+                   filterable?: aggregate.filterable?,
+                   sortable?: aggregate.sortable?,
+                   sensitive?: aggregate.sensitive?,
+                   type: aggregate.type,
+                   include_nil?: aggregate.include_nil?,
+                   constraints: aggregate.constraints,
+                   implementation: aggregate.implementation,
+                   uniq?: aggregate.uniq?,
+                   read_action: read_action,
+                   authorize?: aggregate.authorize?,
+                   join_filters:
+                     Map.new(aggregate.join_filters, &{&1.relationship_path, &1.filter})
+                 ) do
+            query_aggregate = %{query_aggregate | load: aggregate.name}
+
+            case parse_predicates(nested_statement, query_aggregate, context) do
+              {:ok, nested_statement} ->
+                {:ok, BooleanExpression.optimized_new(:and, expression, nested_statement)}
 
-        read_action =
-          aggregate.read_action || Ash.Resource.Info.primary_action!(related, :read).name
-
-        with %{valid?: true} = aggregate_query <- Ash.Query.for_read(related, read_action),
-             %{valid?: true} = aggregate_query <-
-               Ash.Query.Aggregate.build_query(
-                 aggregate_query,
-                 context.resource,
-                 filter: aggregate.filter,
-                 sort: aggregate.sort
-               ),
-             {:ok, query_aggregate} <-
-               Aggregate.new(
-                 context.resource,
-                 aggregate.name,
-                 aggregate.kind,
-                 agg_name: aggregate.name,
-                 path: aggregate.relationship_path,
-                 query: aggregate_query,
-                 field: aggregate.field,
-                 default: aggregate.default,
-                 filterable?: aggregate.filterable?,
-                 sortable?: aggregate.sortable?,
-                 sensitive?: aggregate.sensitive?,
-                 type: aggregate.type,
-                 include_nil?: aggregate.include_nil?,
-                 constraints: aggregate.constraints,
-                 implementation: aggregate.implementation,
-                 uniq?: aggregate.uniq?,
-                 read_action: read_action,
-                 authorize?: aggregate.authorize?,
-                 join_filters: Map.new(aggregate.join_filters, &{&1.relationship_path, &1.filter})
-               ) do
-          query_aggregate = %{query_aggregate | load: aggregate.name}
-
-          case parse_predicates(nested_statement, query_aggregate, context) do
-            {:ok, nested_statement} ->
-              {:ok, BooleanExpression.optimized_new(:and, expression, nested_statement)}
+              {:error, error} ->
+                {:error, error}
+            end
+          else
+            %{valid?: false, errors: errors} ->
+              {:error, errors}
 
             {:error, error} ->
               {:error, error}
           end
         else
-          %{valid?: false, errors: errors} ->
-            {:error, errors}
-
-          {:error, error} ->
-            {:error, error}
+          {:error,
+           Ash.Error.Query.AggregatesNotSupported.exception(
+             resource: context.resource,
+             feature: "using"
+           )}
         end
 
       resource_calculation = calculation(context, field) ->
@@ -3155,31 +3164,36 @@ defmodule Ash.Filter do
 
     opts = Enum.at(args, 1) || []
 
-    if Keyword.keyword?(opts) do
-      kind =
-        if name == :custom_aggregate do
-          :custom
-        else
-          name
-        end
+    kind =
+      if name == :custom_aggregate do
+        :custom
+      else
+        name
+      end
 
-      opts = Keyword.put(opts, :path, path)
+    if Ash.DataLayer.data_layer_can?(context.resource, {:aggregate, kind}) do
+      if Keyword.keyword?(opts) do
+        opts = Keyword.put(opts, :path, path)
 
-      with {:ok, agg} <-
-             Aggregate.new(
-               resource,
-               agg_name(kind, opts),
-               kind,
-               opts
-             ) do
-        {:ok,
-         %Ref{
-           relationship_path: call.relationship_path,
-           attribute: agg
-         }}
+        with {:ok, agg} <-
+               Aggregate.new(
+                 resource,
+                 agg_name(kind, opts),
+                 kind,
+                 opts
+               ) do
+          {:ok,
+           %Ref{
+             relationship_path: call.relationship_path,
+             attribute: agg
+           }}
+        end
+      else
+        {:error, "Aggregate options must be keyword list. In: #{inspect(call)}"}
       end
     else
-      {:error, "Aggregate options must be keyword list. In: #{inspect(call)}"}
+      {:error,
+       Ash.Error.Query.AggregatesNotSupported.exception(resource: resource, feature: "using")}
     end
   end
 
@@ -3467,6 +3481,18 @@ defmodule Ash.Filter do
     end
   end
 
+  def do_hydrate_refs(
+        %Ref{attribute: %Ash.Query.Aggregate{} = agg} = ref,
+        _context
+      ) do
+    if Ash.DataLayer.data_layer_can?(agg.resource, {:aggregate, agg.kind}) do
+      {:ok, ref}
+    else
+      {:error,
+       Ash.Error.Query.AggregatesNotSupported.exception(resource: agg.resource, feature: "using")}
+    end
+  end
+
   def do_hydrate_refs(
         %Ref{attribute: attribute} = ref,
         context
@@ -3498,49 +3524,57 @@ defmodule Ash.Filter do
             end
 
           aggregate = aggregate(context, attribute) ->
-            agg_related = Ash.Resource.Info.related(related, aggregate.relationship_path)
-
-            with %{valid?: true} = aggregate_query <-
-                   Ash.Query.new(agg_related),
-                 %{valid?: true} = aggregate_query <-
-                   Ash.Query.Aggregate.build_query(
-                     aggregate_query,
-                     context.resource,
-                     filter: aggregate.filter,
-                     sort: aggregate.sort
-                   ),
-                 {:ok, query_aggregate} <-
-                   Aggregate.new(
-                     related,
-                     aggregate.name,
-                     aggregate.kind,
-                     agg_name: aggregate.name,
-                     path: aggregate.relationship_path,
-                     query: aggregate_query,
-                     field: aggregate.field,
-                     default: aggregate.default,
-                     filterable?: aggregate.filterable?,
-                     type: aggregate.type,
-                     sortable?: aggregate.sortable?,
-                     include_nil?: aggregate.include_nil?,
-                     sensitive?: aggregate.sensitive?,
-                     constraints: aggregate.constraints,
-                     implementation: aggregate.implementation,
-                     uniq?: aggregate.uniq?,
-                     read_action: aggregate.read_action,
-                     authorize?: aggregate.authorize?,
-                     join_filters:
-                       Map.new(aggregate.join_filters, &{&1.relationship_path, &1.filter})
-                   ) do
-              query_aggregate = %{query_aggregate | load: aggregate.name}
+            if Ash.DataLayer.data_layer_can?(context.resource, {:aggregate, aggregate.kind}) do
+              agg_related = Ash.Resource.Info.related(related, aggregate.relationship_path)
+
+              with %{valid?: true} = aggregate_query <-
+                     Ash.Query.new(agg_related),
+                   %{valid?: true} = aggregate_query <-
+                     Ash.Query.Aggregate.build_query(
+                       aggregate_query,
+                       context.resource,
+                       filter: aggregate.filter,
+                       sort: aggregate.sort
+                     ),
+                   {:ok, query_aggregate} <-
+                     Aggregate.new(
+                       related,
+                       aggregate.name,
+                       aggregate.kind,
+                       agg_name: aggregate.name,
+                       path: aggregate.relationship_path,
+                       query: aggregate_query,
+                       field: aggregate.field,
+                       default: aggregate.default,
+                       filterable?: aggregate.filterable?,
+                       type: aggregate.type,
+                       sortable?: aggregate.sortable?,
+                       include_nil?: aggregate.include_nil?,
+                       sensitive?: aggregate.sensitive?,
+                       constraints: aggregate.constraints,
+                       implementation: aggregate.implementation,
+                       uniq?: aggregate.uniq?,
+                       read_action: aggregate.read_action,
+                       authorize?: aggregate.authorize?,
+                       join_filters:
+                         Map.new(aggregate.join_filters, &{&1.relationship_path, &1.filter})
+                     ) do
+                query_aggregate = %{query_aggregate | load: aggregate.name}
+
+                {:ok, %{ref | attribute: query_aggregate, resource: related}}
+              else
+                %{valid?: false, errors: errors} ->
+                  {:error, errors}
 
-              {:ok, %{ref | attribute: query_aggregate, resource: related}}
+                {:error, error} ->
+                  {:error, error}
+              end
             else
-              %{valid?: false, errors: errors} ->
-                {:error, errors}
-
-              {:error, error} ->
-                {:error, error}
+              {:error,
+               Ash.Error.Query.AggregatesNotSupported.exception(
+                 resource: context.resource,
+                 feature: "using"
+               )}
             end
 
           relationship = relationship(context, attribute) ->
@@ -3566,11 +3600,6 @@ defmodule Ash.Filter do
     end
   end
 
-  def do_hydrate_refs(%Ref{relationship_path: relationship_path, resource: nil} = ref, context) do
-    ref = %{ref | input?: ref.input? || context[:input?] || false}
-    {:ok, %{ref | resource: Ash.Resource.Info.related(context.resource, relationship_path)}}
-  end
-
   def do_hydrate_refs(%BooleanExpression{left: left, right: right} = expr, context) do
     case do_hydrate_refs(left, context) do
       {:ok, true} ->
diff --git a/lib/ash/query/aggregate.ex b/lib/ash/query/aggregate.ex
index 4f6f52f89..b5e4918a7 100644
--- a/lib/ash/query/aggregate.ex
+++ b/lib/ash/query/aggregate.ex
@@ -29,7 +29,7 @@ defmodule Ash.Query.Aggregate do
   @kinds [:count, :first, :sum, :list, :max, :min, :avg, :exists, :custom]
   @type kind :: unquote(Enum.reduce(@kinds, &{:|, [], [&1, &2]}))
 
-  alias Ash.Error.Query.{AggregatesNotSupported, NoReadAction, NoSuchRelationship}
+  alias Ash.Error.Query.{NoReadAction, NoSuchRelationship}
 
   require Ash.Query
 
@@ -190,9 +190,8 @@ defmodule Ash.Query.Aggregate do
           false
       end)
 
-    with {:ok, %Opts{} = opts} <- Opts.validate(opts),
-         agg_name = agg_name(opts),
-         :ok <- validate_supported(resource, kind, agg_name) do
+    with {:ok, %Opts{} = opts} <- Opts.validate(opts) do
+      agg_name = agg_name(opts)
       related = Ash.Resource.Info.related(resource, opts.path)
 
       query =
@@ -395,19 +394,6 @@ defmodule Ash.Query.Aggregate do
     end
   end
 
-  defp validate_supported(resource, kind, nil) do
-    if Ash.DataLayer.data_layer_can?(resource, {:query_aggregate, kind}) do
-      :ok
-    else
-      {:error, AggregatesNotSupported.exception(resource: resource, feature: "using")}
-    end
-  end
-
-  # resource aggregates can only exist if supported, so we don't need to check
-  defp validate_supported(_resource, _kind, _agg_name) do
-    :ok
-  end
-
   defp parse_join_filter(resource, path, filter) do
     [last_relationship | relationships] =
       path_to_reversed_relationships(resource, path)
diff --git a/lib/ash/query/query.ex b/lib/ash/query/query.ex
index e28f21d04..849a4648b 100644
--- a/lib/ash/query/query.ex
+++ b/lib/ash/query/query.ex
@@ -1571,15 +1571,25 @@ defmodule Ash.Query do
         )
 
       match?(%{__struct__: Ash.Query.Aggregate}, field) ->
-        Map.update!(
-          query,
-          :aggregates,
-          &Map.put(
-            &1,
-            field.name,
-            field
+        if Ash.DataLayer.data_layer_can?(query.resource, {:aggregate, field.kind}) do
+          Map.update!(
+            query,
+            :aggregates,
+            &Map.put(
+              &1,
+              field.name,
+              field
+            )
           )
-        )
+        else
+          add_error(
+            query,
+            Ash.Error.Query.AggregatesNotSupported.exception(
+              resource: query.resource,
+              feature: "using"
+            )
+          )
+        end
 
       Ash.Resource.Info.attribute(query.resource, field) ->
         ensure_selected(query, field)