diff --git a/lib/data_layer.ex b/lib/data_layer.ex
index 652ed117..ff42b6e2 100644
--- a/lib/data_layer.ex
+++ b/lib/data_layer.ex
@@ -885,7 +885,8 @@ defmodule AshPostgres.DataLayer do
     functions = [
       AshPostgres.Functions.Like,
       AshPostgres.Functions.ILike,
-      AshPostgres.Functions.Binding
+      AshPostgres.Functions.Binding,
+      AshPostgres.Functions.PostgresIn
     ]
 
     functions =
diff --git a/lib/functions/postgres_in.ex b/lib/functions/postgres_in.ex
new file mode 100644
index 00000000..1e1e26f8
--- /dev/null
+++ b/lib/functions/postgres_in.ex
@@ -0,0 +1,31 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.Functions.PostgresIn do
+  @moduledoc """
+  Generates a native SQL `IN (...)` clause instead of the default `= ANY(...)` array syntax.
+
+  PostgreSQL's query planner may choose different (sometimes suboptimal) indexes when using
+  `= ANY('{...}'::type[])` compared to `IN ($1, $2, ...)`. This function provides an escape
+  hatch for cases where the native `IN` syntax produces better query plans.
+
+  ## Example
+
+      filter(query, postgres_in(id, [^id1, ^id2, ^id3]))
+
+  Generates:
+
+      WHERE id IN ($1, $2, $3)
+
+  Instead of the default:
+
+      WHERE id = ANY($1::uuid[])
+
+  See: https://github.com/ash-project/ash/issues/2605
+  """
+
+  use Ash.Query.Function, name: :postgres_in, predicate?: true
+
+  def args, do: [[:any, {:array, :any}]]
+end
diff --git a/lib/sql_implementation.ex b/lib/sql_implementation.ex
index f66757ae..6cfd59be 100644
--- a/lib/sql_implementation.ex
+++ b/lib/sql_implementation.ex
@@ -177,6 +177,91 @@ defmodule AshPostgres.SqlImplementation do
     {:ok, Ecto.Query.dynamic(fragment("(? <-> ?)", ^arg1, ^arg2)), acc}
   end
 
+  def expr(
+        query,
+        %AshPostgres.Functions.PostgresIn{
+          arguments: [left, right],
+          embedded?: pred_embedded?
+        },
+        bindings,
+        embedded?,
+        acc,
+        _type
+      ) do
+    context_embedded? = pred_embedded? || embedded?
+
+    # Determine the Ecto type from the left-hand side for proper value encoding
+    left_type =
+      case left do
+        %Ash.Query.Ref{attribute: %{type: type, constraints: constraints}} ->
+          AshPostgres.SqlImplementation.parameterized_type(type, constraints)
+
+        _ ->
+          :any
+      end
+
+    {left_expr, acc} =
+      AshSql.Expr.dynamic_expr(query, left, bindings, context_embedded?, :any, acc)
+
+    case right do
+      %Ash.Query.Ref{} ->
+        # If right side is a reference (i.e. an array column), fall back to = ANY(...)
+        {right_expr, acc} =
+          AshSql.Expr.dynamic_expr(query, right, bindings, context_embedded?, :any, acc)
+
+        {:ok, Ecto.Query.dynamic(^left_expr in ^right_expr), acc}
+
+      _ ->
+        values =
+          case right do
+            %Ash.Query.Function.Type{arguments: [value | _]} -> value
+            value -> value
+          end
+
+        values = if is_list(values), do: values, else: [values]
+
+        # Build params and fragment_data in forward order (not reversed)
+        # Param index 0 = left_expr, indices 1..N = values
+        params = [{left_expr, :any}]
+
+        {params, value_fragment_parts, _count, acc} =
+          Enum.reduce(values, {params, [], 1, acc}, fn value, {params, parts, count, acc} ->
+            {value_expr, acc} =
+              AshSql.Expr.dynamic_expr(query, value, bindings, context_embedded?, :any, acc)
+
+            separator =
+              if count == 1, do: "", else: ", "
+
+            typed_value =
+              if left_type != :any do
+                Ecto.Query.dynamic(type(^value_expr, ^left_type))
+              else
+                value_expr
+              end
+
+            new_parts = [{:raw, separator}, {:expr, {:^, [], [count]}}]
+            {params ++ [{typed_value, :any}], parts ++ new_parts, count + 1, acc}
+          end)
+
+        # Build complete fragment: "" left_expr " IN (" v1 ", " v2 ... ")"
+        fragment_data =
+          [{:raw, ""}, {:expr, {:^, [], [0]}}, {:raw, " IN ("}] ++
+            value_fragment_parts ++
+            [{:raw, ")"}]
+
+        dynamic = %Ecto.Query.DynamicExpr{
+          fun: fn _query ->
+            {{:fragment, [], fragment_data}, params, [], %{}}
+          end,
+          binding: [],
+          file: __ENV__.file,
+          line: __ENV__.line
+        }
+
+        {:ok, dynamic, acc}
+    end
+  end
+
   def expr(
         query,
         %Ash.Query.Ref{
diff --git a/test/postgres_in_test.exs b/test/postgres_in_test.exs
new file mode 100644
index 00000000..67efc25c
--- /dev/null
+++ b/test/postgres_in_test.exs
@@ -0,0 +1,79 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.PostgresInTest do
+  use AshPostgres.RepoCase, async: false
+
+  alias AshPostgres.Test.Post
+
+  require Ash.Query
+
+  describe "postgres_in/2" do
+    test "generates SQL IN (...) syntax instead of = ANY(...)" do
+      {query, _vars} =
+        Post
+        |> Ash.Query.filter(postgres_in(id, [^Ash.UUID.generate(), ^Ash.UUID.generate()]))
+        |> Ash.data_layer_query!()
+        |> Map.get(:query)
+        |> then(&AshPostgres.TestRepo.to_sql(:all, &1))
+
+      assert query =~ " IN ("
+      refute query =~ "ANY("
+    end
+
+    test "returns matching records" do
+      post1 =
+        Post
+        |> Ash.Changeset.for_create(:create, %{title: "first"})
+        |> Ash.create!()
+
+      post2 =
+        Post
+        |> Ash.Changeset.for_create(:create, %{title: "second"})
+        |> Ash.create!()
+
+      _post3 =
+        Post
+        |> Ash.Changeset.for_create(:create, %{title: "third"})
+        |> Ash.create!()
+
+      results =
+        Post
+        |> Ash.Query.filter(postgres_in(id, [^post1.id, ^post2.id]))
+        |> Ash.read!()
+        |> Enum.sort_by(& &1.title)
+
+      assert length(results) == 2
+      assert [%{title: "first"}, %{title: "second"}] = results
+    end
+
+    test "returns empty list when no values match" do
+      Post
+      |> Ash.Changeset.for_create(:create, %{title: "existing"})
+      |> Ash.create!()
+
+      results =
+        Post
+        |> Ash.Query.filter(postgres_in(id, [^Ash.UUID.generate()]))
+        |> Ash.read!()
+
+      assert results == []
+    end
+
+    test "works with a single value" do
+      post =
+        Post
+        |> Ash.Changeset.for_create(:create, %{title: "only"})
+        |> Ash.create!()
+
+      results =
+        Post
+        |> Ash.Query.filter(postgres_in(id, [^post.id]))
+        |> Ash.read!()
+
+      assert length(results) == 1
+      assert [%{title: "only"}] = results
+    end
+  end
+end