   Deletes all records that match Ecto.Query
   """
   def execute(_repo, %{sources: {{table, _schema}}, fields: fields, take: _take},
-                      {:nocache, {:delete_all, %Ecto.Query{} = query, limit, context, ordering_fn}},
+                      {:nocache, {:delete_all, %Ecto.Query{} = query, {limit, limit_fn}, context, ordering_fn}},
                       sources, preprocess, opts) do
     context = context |> Context.assign_query(query, sources)
     match_spec = Context.get_match_spec(context)
     preprocess_fn = &process_row(&1, preprocess, fields)
-    Logger.debug("Deleting all records by match specification `#{inspect match_spec}` with limit `#{inspect limit}`")
+    Logger.debug("Deleting all records by match specification `#{inspect match_spec}` with limit #{inspect limit}")
 
     table = table |> Table.get_name()
     Table.transaction(fn ->
       table
-      |> Table.select(match_spec, limit)
+      |> Table.select(match_spec)
       |> Enum.map(fn record ->
         {:ok, _} = Table.delete(table, List.first(record))
         record
       end)
-      |> return_all(ordering_fn, preprocess_fn, opts)
+      |> return_all(ordering_fn, preprocess_fn, {limit, limit_fn}, opts)
     end)
   end
 

          |> preprocess_fn.()
        end)
        |> ordering_fn.()

        {length(result), result}
      _ ->
        {length(records), nil}
    end
  end


  Update all records by a Ecto.Query.
  """
  def execute(_repo, %{sources: {{table, _schema}}, fields: fields, take: _take},
                      {:nocache, {:update_all,
                        %Ecto.Query{updates: updates} = query, {limit, limit_fn}, context, ordering_fn}},
                      sources, preprocess, opts) do
    context = context |> Context.assign_query(query, sources)
    match_spec = Context.get_match_spec(context)
    preprocess_fn = &process_row(&1, preprocess, fields)
    Logger.debug("Updating all records by match specification `#{inspect match_spec}` with limit #{inspect limit}")

    table = table |> Table.get_name()
    Table.transaction(fn ->
      table
      |> Table.select(match_spec)
      |> Enum.map(fn record ->
        update = record
        |> Update.update_record(updates, sources, context)
