Sub query with limit & returning

[Vanja Radovanović]

Hi!

This is what I'd like to be able to express via Ecto:

UPDATE invoices
  SET sync_started_at = $1
  WHERE id IN (
    SELECT id
    FROM invoices
    WHERE
      synced = false
      AND (sync_started_at IS NULL OR sync_started_at < $1)
    LIMIT $2
  )
  RETURNING id::varchar

It is used as a means to prevent multiple app instances from fetching the same set of invoices to process at a given time.

Basically, I need it to mark invoices as "in sync", do that only for "limit" number of invoices for given criteria and return IDs so I can give related workers data to work on.

It could have extracted all (via that "RETURNING") but that's not important for this example.

In our team, we tried several ways to achieve this, but can't get around the "limit" in sub-query.

So, is that possible/how?

If not, would there be interest in supporting this in Ecto or related libs (if so, I'd like to add it)?

Thank you for your time!

Vanja

[José Valim]

Can you post the attempts that you have tried and the errors that you got?

--

José Valim

www.plataformatec.com.br

Skype: jv.ptec

Founder and Director of R&D

[Vanja Radovanović]

Sure

This:

from(i in Invoice,
  where: i.synced == false and
    (is_nil(i.sync_started_at) or i.sync_started_at < ^min_sync_started_at),
  order_by: i.inserted_at,
  limit: ^batch_size
)
|> Repo.update_all(
  set: [sync_started_at: Ecto.DateTime.utc()],
  returning: true
)

Gives:

** (ArgumentError) malformed update `[set: [sync_started_at: #Ecto.DateTime<2017-06-26 13:27:22>], returning: true]` in query expression, expected a keyword list with lists or interpolated expressions as values

        (ecto) lib/ecto/query/builder/update.ex:170: Ecto.Query.Builder.Update.runtime_error!/1

      (elixir) lib/enum.ex:1247: Enum."-map_reduce/3-lists^mapfoldl/2-0-"/3

      (elixir) lib/enum.ex:1247: Enum."-map_reduce/3-lists^mapfoldl/2-0-"/3

        (ecto) lib/ecto/query/builder/update.ex:136: Ecto.Query.Builder.Update.update!/4

        (ecto) lib/ecto/repo/queryable.ex:88: Ecto.Repo.Queryable.update_all/5

Similar one, without "returning":

from(i in Invoice,
  where: i.synced == false and
    (is_nil(i.sync_started_at) or i.sync_started_at < ^min_sync_started_at),
  order_by: i.inserted_at,
  limit: ^batch_size
)
|> Repo.update_all(
  set: [sync_started_at: Ecto.DateTime.utc()]
)

Results in:

** (Ecto.QueryError) `update_all` allows only `where` and `join` expressions in query:

from i in MyApp.Models.Invoice,

  where: i.synced == false and (is_nil(i.sync_started_at) or i.sync_started_at < ^#Ecto.DateTime<2017-06-26 13:30:43>),

  order_by: [asc: i.inserted_at],

  limit: ^10,

  update: [set: [sync_started_at: ^#Ecto.DateTime<2017-06-26 13:31:44>]]

        (ecto) lib/ecto/repo/queryable.ex:122: Ecto.Repo.Queryable.execute/5

Also:

subset_query = from(i in Invoice,
  where: i.synced == false and
    (is_nil(i.sync_started_at) or i.sync_started_at < ^min_sync_started_at),
  order_by: i.inserted_at,
  limit: ^batch_size
)

from(i in subquery(subset_query))
|> Repo.update_all(
  set: [sync_started_at: Ecto.DateTime.utc()]
)

Gives:

** (Ecto.QueryError) `update_all` does not allow subqueries in `from` in query:

from i in subquery(from i in MyApp.Models.Invoice,

  where: i.synced == false and (is_nil(i.sync_started_at) or i.sync_started_at < ^#Ecto.DateTime<2017-06-26 13:40:30>),

  order_by: [asc: i.inserted_at],

  limit: ^10,

  select: i),

  update: [set: [sync_started_at: ^...]]

      (elixir) lib/enum.ex:1623: Enum."-reduce/3-lists^foldl/2-0-"/3

        (ecto) lib/ecto/repo/queryable.ex:122: Ecto.Repo.Queryable.execute/5

I think that's about it with what we've tried, maybe minor variations along the way with the same results.

Hope it helps!

Vanja

[José Valim]

Convert there WHERE into a join. Something along those lines:

subset_query = from(i in Invoice,

  where: i.synced == false and

    (is_nil(i.sync_started_at) or i.sync_started_at < ^min_sync_started_at),

  order_by: i.inserted_at,

  limit: ^batch_size

)

Repo.update_all(

  from(i in Invoice, join: s in subquery(subset_query), on: s.id == i.id)

  [set: [sync_started_at: Ecto.DateTime.utc()]],

  [returning: [:id]]

)

José Valim

www.plataformatec.com.br

Skype: jv.ptec

Founder and Director of R&D

--
You received this message because you are subscribed to the Google Groups "elixir-ecto" group.
To unsubscribe from this group and stop receiving emails from it, send an email to elixir-ecto+unsubscribe@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/elixir-ecto/31ecf86c-da24-400e-b02b-093686704eb1%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Vanja Radovanović]

Yep, that's exactly what did the trick!

Thank you :-)

To unsubscribe from this group and stop receiving emails from it, send an email to elixir-ecto...@googlegroups.com.

[José Valim]

I will add a similar example to the docs because it is a somewhat recurring question.

[Vanja Radovanović]

Just this to examples:?

posts_to_process_query = from(p in Post, select: p.id, where: p.id < 10, order_by: p.inserted_at, limit: 5)
from(p in Post, join: s in subquery(posts_to_process_query), on: s.id == p.id)
|> MyRepo.update_all([set: [processing: true]], [returning: true])

or maybe a separate section with a bit of explanation?

(can add if you can't spare the time)

[José Valim]

Hi Vanja, I have just pushed an example that also includes a bit of explanation. Thanks for the offer though!

José Valim

www.plataformatec.com.br

Skype: jv.ptec

Founder and Director of R&D

[Vanja Radovanović]

Excellent, thank you for solving this and documenting it as well :-)