diff --git a/.tool-versions b/.tool-versions
new file mode 100644
index 0000000..65a8913
--- /dev/null
+++ b/.tool-versions
@@ -0,0 +1,2 @@
+erlang 27.2
+elixir 1.18.1
diff --git a/config/config.exs b/config/config.exs
index d2d855e..becde76 100644
--- a/config/config.exs
+++ b/config/config.exs
@@ -1 +1 @@
-use Mix.Config
+import Config
diff --git a/lib/redshift_ecto.ex b/lib/redshift_ecto.ex
index b6fe516..c6ca676 100644
--- a/lib/redshift_ecto.ex
+++ b/lib/redshift_ecto.ex
@@ -95,7 +95,7 @@ defmodule RedshiftEcto do
   """
 
   # Inherit all behaviour from Ecto.Adapters.SQL
-  use Ecto.Adapters.SQL, :postgrex
+  use Ecto.Adapters.SQL, driver: :postgrex
 
   alias Ecto.Adapters.Postgres
 
@@ -104,6 +104,8 @@ defmodule RedshiftEcto do
   @behaviour Ecto.Adapter.Structure
 
   defdelegate extensions, to: Postgres
+  @impl true
+  defdelegate lock_for_migrations(meta, opts, fun), to: Postgres
 
   ## Custom Redshift types
 
@@ -114,28 +116,76 @@ defmodule RedshiftEcto do
 
   @doc false
   def loaders(:map, type), do: [&json_decode/1, type]
-  def loaders({:map, _}, type), do: [&json_decode/1, type]
+
+  def loaders({:map, _} = type, _) do
+    [&json_decode/1, &Ecto.Type.embedded_load(type, &1, :json)]
+  end
 
   def loaders({:embed, _} = type, _) do
-    [&json_decode/1, &Ecto.Adapters.SQL.load_embed(type, &1)]
+    [&json_decode/1, &Ecto.Type.embedded_load(type, &1, :json)]
   end
 
   def loaders(:binary_id, _type), do: [&{:ok, &1}]
-  def loaders(:uuid, Ecto.UUID), do: [&{:ok, &1}]
+  def loaders(:uuid, _type), do: [&{:ok, &1}]
   def loaders(_, type), do: [type]
 
+  def json_library do
+    case Application.get_env(:postgrex, :json_library) do
+      nil ->
+        default_json_library()
+
+      {library, _opts} ->
+        ensure_json_library(library)
+
+      library ->
+        ensure_json_library(library)
+    end
+  end
+
+  def encode_json!(value) do
+    json_encode!(json_library(), value)
+  end
+
+  defp ensure_json_library(library) do
+    if Code.ensure_loaded?(library) do
+      library
+    else
+      default_json_library()
+    end
+  end
+
+  defp default_json_library do
+    cond do
+      Code.ensure_loaded?(JSON) ->
+        JSON
+
+      Code.ensure_loaded?(Jason) ->
+        Jason
+
+      Code.ensure_loaded?(Poison) ->
+        Poison
+
+      true ->
+        raise "No JSON library configured. Please configure :postgrex, :json_library."
+    end
+  end
+
   defp json_decode(x) when is_binary(x) do
-    {:ok, Ecto.Adapter.json_library().decode!(x)}
+    library = json_library()
+    {:ok, json_decode!(library, x)}
   end
 
   defp json_decode(x), do: {:ok, x}
 
   @doc false
   def dumpers(:map, type), do: [type, &json_encode/1]
-  def dumpers({:map, _}, type), do: [type, &json_encode/1]
+
+  def dumpers({:map, _} = type, _) do
+    [&Ecto.Type.embedded_dump(type, &1, :json), &json_encode/1]
+  end
 
   def dumpers({:embed, _} = type, _) do
-    [&Ecto.Adapters.SQL.dump_embed(type, &1), &json_encode/1]
+    [&Ecto.Type.embedded_dump(type, &1, :json), &json_encode/1]
   end
 
   def dumpers(:binary_id, _type), do: [&Ecto.UUID.cast/1]
@@ -143,14 +193,60 @@ defmodule RedshiftEcto do
   def dumpers(_, type), do: [type]
 
   defp json_encode(%{} = x) do
-    {:ok, Ecto.Adapter.json_library().encode!(x)}
+    library = json_library()
+    {:ok, json_encode!(library, x)}
   end
 
   defp json_encode(x), do: {:ok, x}
 
+  defp json_decode!(library, value) do
+    cond do
+      function_exported?(library, :decode!, 1) ->
+        library.decode!(value)
+
+      function_exported?(library, :decode, 1) ->
+        decode_result(library.decode(value))
+
+      function_exported?(library, :decode, 2) ->
+        decode_result(library.decode(value, []))
+
+      true ->
+        raise ArgumentError,
+              "The configured JSON library #{inspect(library)} does not export decode/1 or decode!/1"
+    end
+  end
+
+  defp decode_result({:ok, decoded}), do: decoded
+  defp decode_result({:ok, decoded, _}), do: decoded
+  defp decode_result(decoded) when is_list(decoded), do: IO.iodata_to_binary(decoded)
+  defp decode_result(decoded), do: decoded
+
+  defp json_encode!(library, value) do
+    cond do
+      function_exported?(library, :encode!, 1) ->
+        library.encode!(value)
+
+      function_exported?(library, :encode, 1) ->
+        encode_result(library.encode(value))
+
+      function_exported?(library, :encode, 2) ->
+        encode_result(library.encode(value, []))
+
+      true ->
+        raise ArgumentError,
+              "The configured JSON library #{inspect(library)} does not export encode/1 or encode!/1"
+    end
+  end
+
+  defp encode_result({:ok, encoded}), do: encode_result(encoded)
+  defp encode_result(encoded) when is_list(encoded), do: IO.iodata_to_binary(encoded)
+  defp encode_result(encoded) when is_binary(encoded), do: encoded
+  defp encode_result(encoded), do: encoded
+
   ## Storage API
 
   @doc false
+  @impl true
   def storage_up(opts) do
     database =
       Keyword.fetch!(opts, :database) || raise ":database is nil in repository configuration"
@@ -173,6 +269,7 @@ defmodule RedshiftEcto do
   end
 
   @doc false
+  @impl true
   def storage_down(opts) do
     database =
       Keyword.fetch!(opts, :database) || raise ":database is nil in repository configuration"
@@ -193,11 +290,36 @@ defmodule RedshiftEcto do
   end
 
   @doc false
+  @impl true
+  def storage_status(opts) do
+    database =
+      Keyword.fetch!(opts, :database) || raise ":database is nil in repository configuration"
+
+    opts = Keyword.put(opts, :database, "template1")
+
+    query =
+      "SELECT datname FROM pg_catalog.pg_database WHERE datname = '#{database}'"
+
+    case run_query(query, opts) do
+      {:ok, %{num_rows: 0}} -> :down
+      {:ok, %{num_rows: _}} -> :up
+      {:error, %{postgres: %{code: :invalid_catalog_name}}} -> :down
+      {:error, error} -> {:error, Exception.message(error)}
+    end
+  end
+
+  @impl true
   def supports_ddl_transaction? do
     true
   end
 
+  @impl true
+  defdelegate dump_cmd(args, opts, config), to: Postgres
+
+  @impl true
   defdelegate structure_dump(default, config), to: Postgres
+
+  @impl true
   defdelegate structure_load(default, config), to: Postgres
 
   ## Helpers
diff --git a/lib/redshift_ecto/connection.ex b/lib/redshift_ecto/connection.ex
index 47c55c4..7b91396 100644
--- a/lib/redshift_ecto/connection.ex
+++ b/lib/redshift_ecto/connection.ex
@@ -9,27 +9,36 @@ if Code.ensure_loaded?(Postgrex) do
 
     ## Module and Options
 
+    @impl true
     def child_spec(opts) do
       opts
       |> Keyword.put_new(:port, @default_port)
-      |> Keyword.put_new(:types, Ecto.Adapters.Postgres.TypeModule)
       |> Postgrex.child_spec()
     end
 
     # constraints may be defined but are not enforced by Amazon Redshift
-    def to_constraints(%Postgrex.Error{}), do: []
+    @impl true
+    def to_constraints(%Postgrex.Error{}, _opts), do: []
 
     ## Query
 
+    @impl true
     defdelegate prepare_execute(conn, name, sql, params, opts), to: Postgres
+    @impl true
     defdelegate execute(conn, sql_or_query, params, opts), to: Postgres
+    @impl true
     defdelegate stream(conn, sql, params, opts), to: Postgres
+    @impl true
+    defdelegate query(conn, sql, params, opts), to: Postgres
+    @impl true
+    defdelegate query_many(conn, sql, params, opts), to: Postgres
 
     alias Ecto.Query
-    alias Ecto.Query.{BooleanExpr, JoinExpr, QueryExpr}
+    alias Ecto.Query.{BooleanExpr, ByExpr, JoinExpr, QueryExpr}
 
-    def all(query) do
-      sources = create_names(query)
+    @impl true
+    def all(query, as_prefix \\ []) do
+      sources = create_names(query, as_prefix)
       {select_distinct, order_by_distinct} = distinct(query.distinct, sources, query)
 
       from = from(query, sources)
@@ -46,6 +55,7 @@ if Code.ensure_loaded?(Postgrex) do
       [select, from, join, where, group_by, having, order_by, limit, offset | lock]
     end
 
+    @impl true
     def update_all(%{from: from, select: nil} = query) do
       sources = sources_unaliased(query)
       {from, _} = get_source(query, sources, 0, from)
@@ -61,6 +71,7 @@ if Code.ensure_loaded?(Postgrex) do
       error!(nil, "RETURNING is not supported by Redshift")
     end
 
+    @impl true
     def update(prefix, table, fields, filters, []) do
       {fields, count} =
         intersperse_reduce(fields, ", ", 1, fn field, acc ->
@@ -79,6 +90,7 @@ if Code.ensure_loaded?(Postgrex) do
       error!(nil, "RETURNING is not supported by Redshift")
     end
 
+    @impl true
     def delete_all(%{from: from, select: nil} = query) do
       sources = sources_unaliased(query)
       {from, _} = get_source(query, sources, 0, from)
@@ -93,6 +105,7 @@ if Code.ensure_loaded?(Postgrex) do
       error!(nil, "RETURNING is not supported by Redshift")
     end
 
+    @impl true
     def delete(prefix, table, filters, []) do
       {filters, _} =
         intersperse_reduce(filters, " AND ", 1, fn field, acc ->
@@ -106,22 +119,35 @@ if Code.ensure_loaded?(Postgrex) do
       error!(nil, "RETURNING is not supported by Redshift")
     end
 
-    def insert(prefix, table, header, rows, {:raise, _, []}, []) do
+    @impl true
+    def explain_query(_conn, _query, _params, _opts) do
+      error!(nil, "EXPLAIN is not supported by Redshift")
+    end
+
+    @impl true
+    def insert(prefix, table, header, rows, {:raise, _, []}, [], placeholders) do
+      counter_offset = length(placeholders) + 1
+
       values =
         if header == [] do
           [" VALUES " | intersperse_map(rows, ?,, fn _ -> "(DEFAULT)" end)]
         else
-          [?\s, ?(, intersperse_map(header, ?,, &quote_name/1), ") VALUES " | insert_all(rows, 1)]
+          [
+            ?\s,
+            ?(,
+            intersperse_map(header, ?,, &quote_name/1),
+            ") VALUES " | insert_all(rows, counter_offset)
+          ]
         end
 
       ["INSERT INTO ", quote_table(prefix, table) | values]
     end
 
-    def insert(_prefix, _table, _header, _rows, _on_conflict, []) do
+    def insert(_prefix, _table, _header, _rows, _on_conflict, [], _placeholders) do
       error!(nil, "ON CONFLICT is not supported by Redshift")
     end
 
-    def insert(_prefix, _table, _header, _rows, _on_conflict, _returning) do
+    def insert(_prefix, _table, _header, _rows, _on_conflict, _returning, _placeholders) do
       error!(nil, "RETURNING is not supported by Redshift")
     end
 
@@ -138,6 +164,9 @@ if Code.ensure_loaded?(Postgrex) do
         nil, counter ->
           {"DEFAULT", counter}
 
+        {:placeholder, placeholder_index}, counter ->
+          {[?$ | placeholder_index], counter}
+
         _, counter ->
           {[?$ | Integer.to_string(counter)], counter + 1}
       end)
@@ -183,11 +212,11 @@ if Code.ensure_loaded?(Postgrex) do
     end
 
     defp distinct(nil, _, _), do: {[], []}
-    defp distinct(%QueryExpr{expr: []}, _, _), do: {[], []}
-    defp distinct(%QueryExpr{expr: true}, _, _), do: {" DISTINCT", []}
-    defp distinct(%QueryExpr{expr: false}, _, _), do: {[], []}
+    defp distinct(%ByExpr{expr: []}, _, _), do: {[], []}
+    defp distinct(%ByExpr{expr: true}, _, _), do: {" DISTINCT", []}
+    defp distinct(%ByExpr{expr: false}, _, _), do: {[], []}
 
-    defp distinct(%QueryExpr{expr: exprs}, sources, query) do
+    defp distinct(%ByExpr{expr: exprs}, sources, query) do
       {[
          " DISTINCT ON (",
          intersperse_map(exprs, ", ", fn {_, expr} -> expr(expr, sources, query) end),
@@ -195,9 +224,9 @@ if Code.ensure_loaded?(Postgrex) do
        ], exprs}
     end
 
-    defp from(%{from: from} = query, sources) do
-      {from, name} = get_source(query, sources, 0, from)
-      [" FROM ", from, " AS " | name]
+    defp from(%{from: %{source: source, hints: hints}} = query, sources) do
+      {from, name} = get_source(query, sources, 0, source)
+      [" FROM ", from, " AS ", name | Enum.map(hints, &[?\s | &1])]
     end
 
     defp update_fields(%Query{updates: updates} = query, sources) do
@@ -287,7 +316,7 @@ if Code.ensure_loaded?(Postgrex) do
     defp group_by(%Query{group_bys: group_bys} = query, sources) do
       [
         " GROUP BY "
-        | intersperse_map(group_bys, ", ", fn %QueryExpr{expr: expr} ->
+        | intersperse_map(group_bys, ", ", fn %ByExpr{expr: expr} ->
             intersperse_map(expr, ", ", &expr(&1, sources, query))
           end)
       ]
@@ -315,7 +344,11 @@ if Code.ensure_loaded?(Postgrex) do
 
     defp limit(%Query{limit: nil}, _sources), do: []
 
-    defp limit(%Query{limit: %QueryExpr{expr: expr}} = query, sources) do
+    defp limit(%Query{limit: %{with_ties: true}} = query, _sources) do
+      error!(query, ":with_ties option is not supported by Redshift")
+    end
+
+    defp limit(%Query{limit: %{expr: expr}} = query, sources) do
       [" LIMIT " | expr(expr, sources, query)]
     end
 
@@ -403,6 +436,10 @@ if Code.ensure_loaded?(Postgrex) do
       all(query)
     end
 
+    defp expr(%Ecto.Query.FromExpr{source: source}, sources, query) do
+      expr(source, sources, query)
+    end
+
     defp expr({:fragment, _, [kw]}, _sources, query) when is_list(kw) or tuple_size(kw) == 3 do
       error!(query, "Redshift adapter does not support keyword or interpolated fragments")
     end
@@ -515,6 +552,10 @@ if Code.ensure_loaded?(Postgrex) do
     defp sources_unaliased(prefix, sources, pos, limit) when pos < limit do
       current =
         case elem(sources, pos) do
+          {table, schema, source_prefix} when table not in [:fragment, :values] ->
+            quoted = quote_table(source_prefix || prefix, table)
+            {quoted, quoted, schema}
+
           {table, schema} ->
             quoted = quote_table(prefix, table)
             {quoted, quoted, schema}
@@ -539,34 +580,51 @@ if Code.ensure_loaded?(Postgrex) do
       []
     end
 
-    defp create_names(%{prefix: prefix, sources: sources}) do
-      create_names(prefix, sources, 0, tuple_size(sources)) |> List.to_tuple()
+    defp create_names(%{prefix: prefix, sources: sources}, as_prefix) do
+      create_names(prefix, sources, 0, tuple_size(sources), as_prefix) |> List.to_tuple()
     end
 
-    defp create_names(prefix, sources, pos, limit) when pos < limit do
-      current =
-        case elem(sources, pos) do
-          {table, schema} ->
-            name = [create_alias(table) | Integer.to_string(pos)]
-            {quote_table(prefix, table), name, schema}
+    defp create_names(prefix, sources, pos, limit, as_prefix) when pos < limit do
+      [
+        create_name(prefix, sources, pos, as_prefix)
+        | create_names(prefix, sources, pos + 1, limit, as_prefix)
+      ]
+    end
 
-          {:fragment, _, _} ->
-            {nil, [?f | Integer.to_string(pos)], nil}
+    defp create_names(_prefix, _sources, pos, pos, as_prefix), do: [as_prefix]
 
-          %Ecto.SubQuery{} ->
-            {nil, [?s | Integer.to_string(pos)], nil}
-        end
+    defp create_name(prefix, sources, pos, as_prefix) do
+      case elem(sources, pos) do
+        {:values, _, _} ->
+          {nil, as_prefix ++ [?v | Integer.to_string(pos)], nil}
 
-      [current | create_names(prefix, sources, pos + 1, limit)]
-    end
+        {:fragment, _, _} ->
+          {nil, as_prefix ++ [?f | Integer.to_string(pos)], nil}
+
+        {table, schema, source_prefix} ->
+          name = as_prefix ++ [create_alias(table) | Integer.to_string(pos)]
+          {quote_table(source_prefix || prefix, table), name, schema}
+
+        {table, schema} ->
+          name = as_prefix ++ [create_alias(table) | Integer.to_string(pos)]
+          {quote_table(prefix, table), name, schema}
 
-    defp create_names(_prefix, _sources, pos, pos), do: []
+        %Ecto.SubQuery{} ->
+          {nil, as_prefix ++ [?s | Integer.to_string(pos)], nil}
+      end
+    end
 
     defp create_alias(<<first, _rest::binary>>) when first in ?a..?z when first in ?A..?Z do
-      <<first>>
+      first
+    end
+
+    defp create_alias(atom) when is_atom(atom) do
+      atom
+      |> Atom.to_string()
+      |> create_alias()
     end
 
-    defp create_alias(_), do: "t"
+    defp create_alias(_), do: ?t
 
     ## DDL
 
@@ -574,6 +632,7 @@ if Code.ensure_loaded?(Postgrex) do
 
     @drops [:drop, :drop_if_exists]
 
+    @impl true
     def execute_ddl({command, %Table{} = table, columns})
         when command in [:create, :create_if_not_exists] do
       table_name = quote_table(table.prefix, table.name)
@@ -671,6 +730,15 @@ if Code.ensure_loaded?(Postgrex) do
     def execute_ddl(keyword) when is_list(keyword),
       do: error!(nil, "PostgreSQL adapter does not support keyword lists in execute")
 
+    @impl true
+    def ddl_logs(_result), do: []
+
+    @impl true
+    def table_exists_query(table) do
+      {"SELECT true FROM information_schema.tables WHERE table_name = $1 AND table_schema = current_schema() LIMIT 1",
+       [table]}
+    end
+
     defp pk_definition(columns, prefix) do
       pks = for {_, name, _, opts} <- columns, opts[:primary_key], do: name
 
@@ -795,7 +863,7 @@ if Code.ensure_loaded?(Postgrex) do
       do: [" DEFAULT ", to_string(literal)]
 
     defp default_expr({:ok, %{} = map}, :map) do
-      default = Ecto.Adapter.json_library().encode!(map)
+      default = RedshiftEcto.encode_json!(map)
       [" DEFAULT ", single_quote(default)]
     end
 
diff --git a/mix.exs b/mix.exs
index a772605..a4d48a2 100644
--- a/mix.exs
+++ b/mix.exs
@@ -7,7 +7,7 @@ defmodule RedshiftEcto.MixProject do
     [
       app: :redshift_ecto,
       version: @version,
-      elixir: "~> 1.6",
+      elixir: "~> 1.18",
       deps: deps(),
       build_per_environment: false,
       build_embedded: Mix.env() == :prod,
@@ -34,11 +34,9 @@ defmodule RedshiftEcto.MixProject do
 
   defp deps do
     [
-      {:ecto, "~> 2.2"},
-      {:postgrex, "~> 0.13"},
-      {:ecto_replay_sandbox, "~> 1.0.0"},
-      {:ex_doc, "~> 0.18", only: :dev, runtime: false},
-      {:poison, "~> 2.2 or ~> 3.0", optional: true}
+      {:ecto, "~> 3.13", override: true},
+      {:ecto_sql, "~> 3.13"},
+      {:postgrex, "~> 0.21"}
     ]
   end
 
diff --git a/mix.lock b/mix.lock
index 512c4eb..2baab9b 100644
--- a/mix.lock
+++ b/mix.lock
@@ -1,12 +1,38 @@
 %{
-  "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm"},
-  "db_connection": {:hex, :db_connection, "1.1.3", "89b30ca1ef0a3b469b1c779579590688561d586694a3ce8792985d4d7e575a61", [:mix], [{:connection, "~> 1.0.2", [hex: :connection, repo: "hexpm", optional: false]}, {:poolboy, "~> 1.5", [hex: :poolboy, repo: "hexpm", optional: true]}, {:sbroker, "~> 1.0", [hex: :sbroker, repo: "hexpm", optional: true]}], "hexpm"},
-  "decimal": {:hex, :decimal, "1.5.0", "b0433a36d0e2430e3d50291b1c65f53c37d56f83665b43d79963684865beab68", [:mix], [], "hexpm"},
-  "earmark": {:hex, :earmark, "1.2.4", "99b637c62a4d65a20a9fb674b8cffb8baa771c04605a80c911c4418c69b75439", [:mix], [], "hexpm"},
-  "ecto": {:hex, :ecto, "2.2.9", "031d55df9bb430cb118e6f3026a87408d9ce9638737bda3871e5d727a3594aae", [:mix], [{:db_connection, "~> 1.1", [hex: :db_connection, repo: "hexpm", optional: true]}, {:decimal, "~> 1.2", [hex: :decimal, repo: "hexpm", optional: false]}, {:mariaex, "~> 0.8.0", [hex: :mariaex, repo: "hexpm", optional: true]}, {:poison, "~> 2.2 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: true]}, {:poolboy, "~> 1.5", [hex: :poolboy, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.13.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:sbroker, "~> 1.0", [hex: :sbroker, repo: "hexpm", optional: true]}], "hexpm"},
-  "ecto_replay_sandbox": {:hex, :ecto_replay_sandbox, "1.0.0", "41c68eee561d811e1cc7f4f3e02490f76503f0e8dd4984fa2f5d1f4580787a37", [:mix], [{:db_connection, "~> 1.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 2.2", [hex: :ecto, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.13", [hex: :postgrex, repo: "hexpm", optional: false]}], "hexpm"},
-  "ex_doc": {:hex, :ex_doc, "0.18.3", "f4b0e4a2ec6f333dccf761838a4b253d75e11f714b85ae271c9ae361367897b7", [:mix], [{:earmark, "~> 1.1", [hex: :earmark, repo: "hexpm", optional: false]}], "hexpm"},
-  "poison": {:hex, :poison, "3.1.0", "d9eb636610e096f86f25d9a46f35a9facac35609a7591b3be3326e99a0484665", [:mix], [], "hexpm"},
-  "poolboy": {:hex, :poolboy, "1.5.1", "6b46163901cfd0a1b43d692657ed9d7e599853b3b21b95ae5ae0a777cf9b6ca8", [:rebar], [], "hexpm"},
-  "postgrex": {:hex, :postgrex, "0.13.5", "3d931aba29363e1443da167a4b12f06dcd171103c424de15e5f3fc2ba3e6d9c5", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 1.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.0", [hex: :decimal, repo: "hexpm", optional: false]}], "hexpm"},
+  "db_connection":
+    {:hex, :db_connection, "2.8.1", "9abdc1e68c34c6163f6fb96a96532272d13ad7ca45262156ae8b7ec6d9dc4bec", [:mix],
+     [
+       {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}
+     ], "hexpm", "a61a3d489b239d76f326e03b98794fb8e45168396c925ef25feb405ed09da8fd"},
+  "decimal":
+    {:hex, :decimal, "2.3.0", "3ad6255aa77b4a3c4f818171b12d237500e63525c2fd056699967a3e7ea20f62", [:mix], [], "hexpm",
+     "a4d66355cb29cb47c3cf30e71329e58361cfcb37c34235ef3bf1d7bf3773aeac"},
+  "ecto":
+    {:hex, :ecto, "3.13.2", "7d0c0863f3fc8d71d17fc3ad3b9424beae13f02712ad84191a826c7169484f01", [:mix],
+     [
+       {:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]},
+       {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]},
+       {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}
+     ], "hexpm", "669d9291370513ff56e7b7e7081b7af3283d02e046cf3d403053c557894a0b3e"},
+  "ecto_sql":
+    {:hex, :ecto_sql, "3.13.2", "a07d2461d84107b3d037097c822ffdd36ed69d1cf7c0f70e12a3d1decf04e2e1", [:mix],
+     [
+       {:db_connection, "~> 2.5 or ~> 2.4.1", [hex: :db_connection, repo: "hexpm", optional: false]},
+       {:ecto, "~> 3.13.0", [hex: :ecto, repo: "hexpm", optional: false]},
+       {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]},
+       {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]},
+       {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]},
+       {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}
+     ], "hexpm", "539274ab0ecf1a0078a6a72ef3465629e4d6018a3028095dc90f60a19c371717"},
+  "postgrex":
+    {:hex, :postgrex, "0.21.1", "2c5cc830ec11e7a0067dd4d623c049b3ef807e9507a424985b8dcf921224cd88", [:mix],
+     [
+       {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]},
+       {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]},
+       {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]},
+       {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}
+     ], "hexpm", "27d8d21c103c3cc68851b533ff99eef353e6a0ff98dc444ea751de43eb48bdac"},
+  "telemetry":
+    {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm",
+     "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"}
 }
diff --git a/test/redshift_ecto_test.exs b/test/redshift_ecto_test.exs
index dfc450d..1266327 100644
--- a/test/redshift_ecto_test.exs
+++ b/test/redshift_ecto_test.exs
@@ -54,7 +54,7 @@ defmodule RedshiftEctoTest do
   end
 
   defp normalize(query, operation \\ :all, counter \\ 0) do
-    {query, _params, _key} = Ecto.Query.Planner.prepare(query, operation, RedshiftEcto, counter)
+    {query, _params, _key} = Ecto.Query.Planner.plan(query, operation, RedshiftEcto)
     {query, _} = Ecto.Query.Planner.normalize(query, operation, RedshiftEcto, counter)
     query
   end
@@ -64,8 +64,10 @@ defmodule RedshiftEctoTest do
   defp delete_all(query), do: query |> SQL.delete_all() |> IO.iodata_to_binary()
   defp execute_ddl(query), do: query |> SQL.execute_ddl() |> Enum.map(&IO.iodata_to_binary/1)
 
-  defp insert(prefx, table, header, rows, on_conflict, returning) do
-    IO.iodata_to_binary(SQL.insert(prefx, table, header, rows, on_conflict, returning))
+  defp insert(prefx, table, header, rows, on_conflict, returning, placeholders \\ []) do
+    IO.iodata_to_binary(
+      SQL.insert(prefx, table, header, rows, on_conflict, returning, placeholders)
+    )
   end
 
   defp update(prefx, table, fields, filter, returning) do
@@ -94,7 +96,7 @@ defmodule RedshiftEctoTest do
     assert_raise Ecto.QueryError,
                  ~r"Redshift does not support selecting all fields from \"posts\" without a schema",
                  fn ->
-                   all from(p in "posts", select: p) |> normalize()
+                   all(from(p in "posts", select: p) |> normalize())
                  end
   end
 
@@ -309,7 +311,7 @@ defmodule RedshiftEctoTest do
 
     assert all(query) == ~s{SELECT $1::char(36) FROM "schema" AS s0}
 
-    query = Schema |> select([], type(^1, Custom.Permalink)) |> normalize
+    query = Schema |> select([], type(^1, CustomPermalink)) |> normalize
     assert all(query) == ~s{SELECT $1::bigint FROM "schema" AS s0}
   end
 
@@ -405,8 +407,8 @@ defmodule RedshiftEctoTest do
     query =
       "schema"
       |> select([m], {m.id, ^true})
-      |> join(:inner, [], Schema2, fragment("?", ^true))
-      |> join(:inner, [], Schema2, fragment("?", ^false))
+      |> join(:inner, [], Schema2, on: fragment("?", ^true))
+      |> join(:inner, [], Schema2, on: fragment("?", ^false))
       |> where([], fragment("?", ^true))
       |> where([], fragment("?", ^false))
       |> having([], fragment("?", ^true))
@@ -481,7 +483,7 @@ defmodule RedshiftEctoTest do
 
     query =
       Schema
-      |> join(:inner, [p], q in Schema2, p.x == q.z)
+      |> join(:inner, [p], q in Schema2, on: p.x == q.z)
       |> update([_], set: [x: 0])
       |> normalize(:update_all)
 
@@ -523,7 +525,7 @@ defmodule RedshiftEctoTest do
     query = from(e in Schema, where: e.x == 123) |> normalize
     assert delete_all(query) == ~s{DELETE FROM "schema" WHERE ("schema"."x" = 123)}
 
-    query = Schema |> join(:inner, [p], q in Schema2, p.x == q.z) |> normalize
+    query = Schema |> join(:inner, [p], q in Schema2, on: p.x == q.z) |> normalize
 
     assert delete_all(query) ==
              ~s{DELETE FROM "schema" USING "schema2" WHERE ("schema"."x" = "schema2"."z")}
@@ -557,15 +559,16 @@ defmodule RedshiftEctoTest do
   ## Joins
 
   test "join" do
-    query = Schema |> join(:inner, [p], q in Schema2, p.x == q.z) |> select([], true) |> normalize
+    query =
+      Schema |> join(:inner, [p], q in Schema2, on: p.x == q.z) |> select([], true) |> normalize
 
     assert all(query) ==
              ~s{SELECT TRUE FROM "schema" AS s0 INNER JOIN "schema2" AS s1 ON s0."x" = s1."z"}
 
     query =
       Schema
-      |> join(:inner, [p], q in Schema2, p.x == q.z)
-      |> join(:inner, [], Schema, true)
+      |> join(:inner, [p], q in Schema2, on: p.x == q.z)
+      |> join(:inner, [], Schema, on: true)
       |> select([], true)
       |> normalize
 
@@ -575,7 +578,8 @@ defmodule RedshiftEctoTest do
   end
 
   test "join with nothing bound" do
-    query = Schema |> join(:inner, [], q in Schema2, q.z == q.z) |> select([], true) |> normalize
+    query =
+      Schema |> join(:inner, [], q in Schema2, on: q.z == q.z) |> select([], true) |> normalize
 
     assert all(query) ==
              ~s{SELECT TRUE FROM "schema" AS s0 INNER JOIN "schema2" AS s1 ON s1."z" = s1."z"}
@@ -583,7 +587,10 @@ defmodule RedshiftEctoTest do
 
   test "join without schema" do
     query =
-      "posts" |> join(:inner, [p], q in "comments", p.x == q.z) |> select([], true) |> normalize
+      "posts"
+      |> join(:inner, [p], q in "comments", on: p.x == q.z)
+      |> select([], true)
+      |> normalize
 
     assert all(query) ==
              ~s{SELECT TRUE FROM "posts" AS p0 INNER JOIN "comments" AS c1 ON p0."x" = c1."z"}
@@ -594,7 +601,7 @@ defmodule RedshiftEctoTest do
 
     query =
       "comments"
-      |> join(:inner, [c], p in subquery(posts), true)
+      |> join(:inner, [c], p in subquery(posts), on: true)
       |> select([_, p], p.x)
       |> normalize
 
@@ -606,7 +613,7 @@ defmodule RedshiftEctoTest do
 
     query =
       "comments"
-      |> join(:inner, [c], p in subquery(posts), true)
+      |> join(:inner, [c], p in subquery(posts), on: true)
       |> select([_, p], p)
       |> normalize
 
@@ -616,7 +623,8 @@ defmodule RedshiftEctoTest do
   end
 
   test "join with prefix" do
-    query = Schema |> join(:inner, [p], q in Schema2, p.x == q.z) |> select([], true) |> normalize
+    query =
+      Schema |> join(:inner, [p], q in Schema2, on: p.x == q.z) |> select([], true) |> normalize
 
     assert all(%{query | prefix: "prefix"}) ==
              ~s{SELECT TRUE FROM "prefix"."schema" AS s0 INNER JOIN "prefix"."schema2" AS s1 ON s0."x" = s1."z"}
@@ -628,7 +636,8 @@ defmodule RedshiftEctoTest do
       |> join(
         :inner,
         [p],
-        q in fragment("SELECT * FROM schema2 AS s2 WHERE s2.id = ? AND s2.field = ?", p.x, ^10)
+        q in fragment("SELECT * FROM schema2 AS s2 WHERE s2.id = ? AND s2.field = ?", p.x, ^10),
+        on: true
       )
       |> select([p], {p.id, ^0})
       |> where([p], p.id > 0 and p.id < ^100)
@@ -643,7 +652,7 @@ defmodule RedshiftEctoTest do
   test "join with fragment and on defined" do
     query =
       Schema
-      |> join(:inner, [p], q in fragment("SELECT * FROM schema2"), q.id == p.id)
+      |> join(:inner, [p], q in fragment("SELECT * FROM schema2"), on: q.id == p.id)
       |> select([p], {p.id, ^0})
       |> normalize
 
@@ -654,7 +663,9 @@ defmodule RedshiftEctoTest do
 
   test "join with query interpolation" do
     inner = Ecto.Queryable.to_query(Schema2)
-    query = from(p in Schema, left_join: c in ^inner, select: {p.id, c.id}) |> normalize()
+
+    query =
+      from(p in Schema, left_join: c in ^inner, on: true, select: {p.id, c.id}) |> normalize()
 
     assert all(query) ==
              "SELECT s0.\"id\", s1.\"id\" FROM \"schema\" AS s0 LEFT OUTER JOIN \"schema2\" AS s1 ON TRUE"
@@ -666,7 +677,8 @@ defmodule RedshiftEctoTest do
       |> join(
         :inner_lateral,
         [p],
-        q in fragment("SELECT * FROM schema2 AS s2 WHERE s2.id = ? AND s2.field = ?", p.x, ^10)
+        q in fragment("SELECT * FROM schema2 AS s2 WHERE s2.id = ? AND s2.field = ?", p.x, ^10),
+        on: true
       )
       |> select([p, q], {p.id, q.z})
       |> where([p], p.id > 0 and p.id < ^100)
@@ -697,7 +709,7 @@ defmodule RedshiftEctoTest do
   describe "query interpolation parameters" do
     test "self join on subquery" do
       subquery = select(Schema, [r], %{x: r.x, y: r.y})
-      query = subquery |> join(:inner, [c], p in subquery(subquery), true) |> normalize
+      query = subquery |> join(:inner, [c], p in subquery(subquery), on: true) |> normalize
 
       assert all(query) ==
                ~s{SELECT s0."x", s0."y" FROM "schema" AS s0 INNER JOIN } <>
@@ -707,7 +719,7 @@ defmodule RedshiftEctoTest do
 
     test "self join on subquery with fragment" do
       subquery = select(Schema, [r], %{string: fragment("downcase(?)", ^"string")})
-      query = subquery |> join(:inner, [c], p in subquery(subquery), true) |> normalize
+      query = subquery |> join(:inner, [c], p in subquery(subquery), on: true) |> normalize
 
       assert all(query) ==
                ~s{SELECT downcase($1) FROM "schema" AS s0 INNER JOIN } <>
@@ -720,7 +732,7 @@ defmodule RedshiftEctoTest do
       query =
         Schema
         |> select([r], %{y: ^666})
-        |> join(:inner, [c], p in subquery(subquery), true)
+        |> join(:inner, [c], p in subquery(subquery), on: true)
         |> where([a, b], a.x == ^111)
         |> normalize