diff --git a/lib/sql_implementation.ex b/lib/sql_implementation.ex
index 4029474..7324f38 100644
--- a/lib/sql_implementation.ex
+++ b/lib/sql_implementation.ex
@@ -24,74 +24,77 @@ defmodule AshSqlite.SqlImplementation do
   @impl true
   def expr(
         query,
-        %like{arguments: [arg1, arg2], embedded?: pred_embedded?},
+        %Ash.Query.Operator.In{
+          right: %Ash.Query.Function.Type{arguments: [right | _]}
+        } = op,
         bindings,
         embedded?,
         acc,
         type
       )
-      when like in [AshSqlite.Functions.Like, AshSqlite.Functions.ILike] do
-    {arg1, acc} =
-      AshSql.Expr.dynamic_expr(query, arg1, bindings, pred_embedded? || embedded?, :string, acc)
-
-    {arg2, acc} =
-      AshSql.Expr.dynamic_expr(query, arg2, bindings, pred_embedded? || embedded?, :string, acc)
-
-    inner_dyn =
-      if like == AshSqlite.Functions.Like do
-        Ecto.Query.dynamic(like(^arg1, ^arg2))
-      else
-        Ecto.Query.dynamic(like(fragment("LOWER(?)", ^arg1), fragment("LOWER(?)", ^arg2)))
-      end
-
-    if type != Ash.Type.Boolean do
-      {:ok, inner_dyn, acc}
-    else
-      {:ok, Ecto.Query.dynamic(type(^inner_dyn, ^type)), acc}
-    end
+      when is_list(right) or is_struct(right, MapSet) do
+    expr(query, %{op | right: right}, bindings, embedded?, acc, type)
   end
 
   def expr(
         query,
-        %Ash.Query.Operator.In{
-          right: %Ash.Query.Function.Type{arguments: [right | _]} = type
-        } = op,
+        %Ash.Query.Operator.In{left: left, right: right, embedded?: pred_embedded?},
         bindings,
         embedded?,
         acc,
-        type
+        _type
       )
       when is_list(right) or is_struct(right, MapSet) do
-    expr(query, %{op | right: right}, bindings, embedded?, acc, type)
+    {item_type, constraints} = in_item_type(left)
+    context_embedded? = pred_embedded? || embedded?
+    values = Enum.to_list(right)
+
+    if Enum.any?(values, &complex_in_value?/1) do
+      expand_in_to_or(query, left, values, bindings, context_embedded?, acc, item_type)
+    else
+      {left_expr, acc} =
+        AshSql.Expr.dynamic_expr(
+          query,
+          left,
+          in_left_bindings(bindings, item_type, constraints),
+          context_embedded?,
+          in_left_type(item_type, constraints),
+          acc
+        )
+
+      values = dump_in_values(query, bindings, values, item_type, constraints)
+
+      {:ok, Ecto.Query.dynamic(^left_expr in ^values), acc}
+    end
   end
 
   def expr(
         query,
-        %Ash.Query.Operator.In{left: left, right: right, embedded?: pred_embedded?},
+        %like{arguments: [arg1, arg2], embedded?: pred_embedded?},
         bindings,
         embedded?,
         acc,
         type
       )
-      when is_list(right) or is_struct(right, MapSet) do
-    right
-    |> Enum.reduce(nil, fn val, acc ->
-      if is_nil(acc) do
-        %Ash.Query.Operator.Eq{left: left, right: val}
+      when like in [AshSqlite.Functions.Like, AshSqlite.Functions.ILike] do
+    {arg1, acc} =
+      AshSql.Expr.dynamic_expr(query, arg1, bindings, pred_embedded? || embedded?, :string, acc)
+
+    {arg2, acc} =
+      AshSql.Expr.dynamic_expr(query, arg2, bindings, pred_embedded? || embedded?, :string, acc)
+
+    inner_dyn =
+      if like == AshSqlite.Functions.Like do
+        Ecto.Query.dynamic(like(^arg1, ^arg2))
       else
-        %Ash.Query.BooleanExpression{
-          op: :or,
-          left: acc,
-          right: %Ash.Query.Operator.Eq{left: left, right: val}
-        }
+        Ecto.Query.dynamic(like(fragment("LOWER(?)", ^arg1), fragment("LOWER(?)", ^arg2)))
       end
-    end)
-    |> then(fn expr ->
-      {expr, acc} =
-        AshSql.Expr.dynamic_expr(query, expr, bindings, pred_embedded? || embedded?, type, acc)
 
-      {:ok, expr, acc}
-    end)
+    if type != Ash.Type.Boolean do
+      {:ok, inner_dyn, acc}
+    else
+      {:ok, Ecto.Query.dynamic(type(^inner_dyn, ^type)), acc}
+    end
   end
 
   def expr(
@@ -289,6 +292,120 @@ defmodule AshSqlite.SqlImplementation do
   defp plain_map?(value) when is_map(value) and not is_struct(value), do: true
   defp plain_map?(_), do: false
 
+  defp in_item_type(%Ash.Query.Ref{attribute: %{type: type} = attribute}) do
+    {type, Map.get(attribute, :constraints, []) || []}
+  end
+
+  defp in_item_type(_), do: {nil, []}
+
+  defp in_left_bindings(bindings, item_type, constraints) do
+    if ci_string_type?(item_type, constraints) do
+      bindings
+    else
+      Map.put(bindings, :no_cast?, true)
+    end
+  end
+
+  defp in_left_type(item_type, constraints) do
+    if ci_string_type?(item_type, constraints) do
+      if constraints == [] do
+        item_type
+      else
+        {item_type, constraints}
+      end
+    end
+  end
+
+  defp complex_in_value?(value) do
+    Ash.Expr.expr?(value) || is_list(value) || (is_map(value) && !is_struct(value))
+  end
+
+  defp expand_in_to_or(query, left, values, bindings, embedded?, acc, type) do
+    values
+    |> Enum.reduce(nil, fn value, acc ->
+      if is_nil(acc) do
+        %Ash.Query.Operator.Eq{left: left, right: value}
+      else
+        %Ash.Query.BooleanExpression{
+          op: :or,
+          left: acc,
+          right: %Ash.Query.Operator.Eq{left: left, right: value}
+        }
+      end
+    end)
+    |> then(fn expr ->
+      {expr, acc} = AshSql.Expr.dynamic_expr(query, expr, bindings, embedded?, type, acc)
+      {:ok, expr, acc}
+    end)
+  end
+
+  defp dump_in_values(_query, _bindings, values, nil, _constraints) do
+    Enum.map(values, fn
+      # Preserve the old equality fallback for untyped atom values when the LHS has no attribute.
+      value when is_atom(value) and not is_boolean(value) and not is_nil(value) ->
+        to_string(value)
+
+      value ->
+        value
+    end)
+  end
+
+  defp dump_in_values(query, bindings, values, item_type, constraints) do
+    ecto_type =
+      parameterized_type(item_type, constraints) ||
+        item_type
+        |> Ash.Type.get_type()
+        |> Ash.Type.storage_type(constraints)
+
+    adapter = sqlite_adapter!(query, bindings)
+
+    Enum.map(values, fn value ->
+      case Ecto.Type.adapter_dump(adapter, ecto_type, value) do
+        {:ok, value} -> value
+        # Some custom/already-dumped values may not accept another dump; keep the old value.
+        _ -> value
+      end
+    end)
+  end
+
+  defp sqlite_adapter!(query, bindings) do
+    repo =
+      bindings
+      |> Map.fetch!(:resource)
+      |> AshSql.dynamic_repo(__MODULE__, query)
+
+    case repo.__adapter__() do
+      Ecto.Adapters.SQLite3 ->
+        Ecto.Adapters.SQLite3
+
+      adapter ->
+        raise ArgumentError,
+              "expected #{inspect(repo)} to use sqlite adapter `Ecto.Adapters.SQLite3`, got: #{inspect(adapter)}"
+    end
+  end
+
+  defp ci_string_type?({:parameterized, {inner_type, constraints}}, []) do
+    parameterized_ci_string_type?(inner_type, constraints)
+  end
+
+  defp ci_string_type?({:parameterized, inner_type, constraints}, []) do
+    parameterized_ci_string_type?(inner_type, constraints)
+  end
+
+  defp ci_string_type?(type, constraints) when is_atom(type) do
+    type = Ash.Type.get_type(type)
+    Ash.Type.ash_type?(type) && Ash.Type.storage_type(type, constraints) == :ci_string
+  end
+
+  defp ci_string_type?(_, _), do: false
+
+  defp parameterized_ci_string_type?(inner_type, constraints)
+       when is_atom(inner_type) and is_list(constraints) do
+    function_exported?(inner_type, :type, 1) && inner_type.type(constraints) == :ci_string
+  end
+
+  defp parameterized_ci_string_type?(_, _), do: false
+
   @impl true
   def type_expr(expr, nil), do: expr
 
diff --git a/test/filter_test.exs b/test/filter_test.exs
index dc21832..2842dd0 100644
--- a/test/filter_test.exs
+++ b/test/filter_test.exs
@@ -4,7 +4,7 @@
 
 defmodule AshSqlite.FilterTest do
   use AshSqlite.RepoCase, async: false
-  alias AshSqlite.Test.{Author, Comment, Post}
+  alias AshSqlite.Test.{Author, Comment, IntegerPost, Post}
 
   require Ash.Query
 
@@ -183,6 +183,134 @@ defmodule AshSqlite.FilterTest do
                |> Ash.Query.sort(title: :asc)
                |> Ash.read!()
     end
+
+    test "large pinned lists do not compile to nested OR expressions" do
+      titles = Enum.map(1..1100, &"title#{&1}")
+
+      query =
+        Post
+        |> Ash.Query.new()
+        |> Ash.Query.filter(title in ^titles)
+
+      {:ok, ecto_query} = Ash.Query.data_layer_query(query)
+      {sql, _params} = Ecto.Adapters.SQL.to_sql(:all, TestRepo, ecto_query)
+
+      assert sql =~ " IN "
+      assert sql =~ ~s(p0."title" IN)
+      refute sql =~ " OR "
+      refute sql =~ ~S|CAST(p0."title" AS TEXT) IN|
+
+      assert [] = Ash.read!(query)
+    end
+
+    test "empty pinned lists return no rows" do
+      assert [] =
+               Post
+               |> Ash.Query.filter(id in ^[])
+               |> Ash.read!()
+    end
+
+    test "complex list values keep the existing OR fallback" do
+      query =
+        Post
+        |> Ash.Query.new()
+        |> Ash.Query.filter(stuff in ^[%{"kind" => "one"}, %{"kind" => "two"}])
+
+      {:ok, ecto_query} = Ash.Query.data_layer_query(query)
+      {sql, _params} = Ecto.Adapters.SQL.to_sql(:all, TestRepo, ecto_query)
+
+      assert sql =~ " OR "
+      assert sql =~ ~S|json(p0."stuff")|
+      assert [] = Ash.read!(query)
+    end
+
+    test "it properly filters typed scalar values" do
+      post_id = Ash.UUID.generate()
+
+      post =
+        Post
+        |> Ash.Changeset.for_create(:create, %{
+          id: post_id,
+          title: "typed",
+          score: 7,
+          public: true,
+          category: "MiXeD",
+          status: :open,
+          status_enum: :closed
+        })
+        |> Ash.create!()
+
+      no_cast_post =
+        Post
+        |> Ash.Changeset.for_create(:create, %{title: "no cast", status_enum_no_cast: :open})
+        |> Ash.create!()
+
+      integer_post =
+        IntegerPost
+        |> Ash.Changeset.for_create(:create, %{title: "integer"})
+        |> Ash.create!()
+
+      no_cast_post_id = no_cast_post.id
+      integer_post_id = integer_post.id
+
+      score_query =
+        Post
+        |> Ash.Query.filter(score in ^[7])
+
+      {:ok, ecto_query} = Ash.Query.data_layer_query(score_query)
+      {sql, _params} = Ecto.Adapters.SQL.to_sql(:all, TestRepo, ecto_query)
+
+      assert sql =~ ~s(p0."score" IN)
+      refute sql =~ ~S|CAST(p0."score" AS INTEGER) IN|
+
+      assert [%Post{id: ^post_id}] =
+               Post
+               |> Ash.Query.filter(id in ^[post_id])
+               |> Ash.read!()
+
+      assert [%Post{id: ^post_id}] =
+               Ash.read!(score_query)
+
+      assert [%Post{id: ^post_id}] =
+               Post
+               |> Ash.Query.filter(public in ^[true] and title in ^["typed"])
+               |> Ash.read!()
+
+      assert [%Post{id: ^post_id}] =
+               Post
+               |> Ash.Query.filter(category in ^["mixed"])
+               |> Ash.read!()
+
+      assert [%Post{id: ^post_id}] =
+               Post
+               |> Ash.Query.filter(status in ^[:open] and title in ^["typed"])
+               |> Ash.read!()
+
+      assert [%Post{id: ^post_id}] =
+               Post
+               |> Ash.Query.filter(status_enum in ^[:closed] and title in ^["typed"])
+               |> Ash.read!()
+
+      assert [%Post{id: ^post_id}] =
+               Post
+               |> Ash.Query.filter(created_at in ^[post.created_at] and title in ^["typed"])
+               |> Ash.read!()
+
+      assert [%Post{id: ^post_id}] =
+               Post
+               |> Ash.Query.filter(decimal in ^[Decimal.new("0")] and title in ^["typed"])
+               |> Ash.read!()
+
+      assert [%Post{id: ^no_cast_post_id}] =
+               Post
+               |> Ash.Query.filter(status_enum_no_cast in ^[:open] and title in ^["no cast"])
+               |> Ash.read!()
+
+      assert [%IntegerPost{id: ^integer_post_id}] =
+               IntegerPost
+               |> Ash.Query.filter(id in ^[integer_post_id])
+               |> Ash.read!()
+    end
   end
 
   describe "with a boolean filter applied" do