Current section
20 Versions
Jump to
Current section
20 Versions
Compare versions
21
files changed
+620
additions
-86
deletions
| @@ -0,0 +1,15 @@ | |
| 1 | + # Changelog for v0.x |
| 2 | + |
| 3 | + ## v0.2.0 (2024-09-21) |
| 4 | + |
| 5 | + ### Bug fixes |
| 6 | + |
| 7 | + ### Enhancements |
| 8 | + |
| 9 | + * Upserts: Support for Ecto options `:on_conflict` and `:conflict_target` |
| 10 | + * Pipelining: New Repo functions for async/await within a transaction. |
| 11 | + |
| 12 | + ### New Documentation |
| 13 | + |
| 14 | + * TESTING.md: Document to describe how to set up a Sandbox |
| 15 | + * CHANGELOG.md: This file! |
| @@ -2,7 +2,7 @@ | |
| 2 2 | [{<<"GitHub">>, |
| 3 3 | <<"https://github.com/foundationdb-beam/ecto_foundationdb">>}]}. |
| 4 4 | {<<"name">>,<<"ecto_foundationdb">>}. |
| 5 | - {<<"version">>,<<"0.1.2">>}. |
| 5 | + {<<"version">>,<<"0.2.0">>}. |
| 6 6 | {<<"description">>,<<"FoundationDB adapter for Ecto">>}. |
| 7 7 | {<<"elixir">>,<<"~> 1.15">>}. |
| 8 8 | {<<"app">>,<<"ecto_foundationdb">>}. |
| @@ -11,7 +11,7 @@ | |
| 11 11 | [[{<<"name">>,<<"erlfdb">>}, |
| 12 12 | {<<"app">>,<<"erlfdb">>}, |
| 13 13 | {<<"optional">>,false}, |
| 14 | - {<<"requirement">>,<<"~> 0.1">>}, |
| 14 | + {<<"requirement">>,<<"~> 0.2">>}, |
| 15 15 | {<<"repository">>,<<"hexpm">>}], |
| 16 16 | [{<<"name">>,<<"ecto">>}, |
| 17 17 | {<<"app">>,<<"ecto">>}, |
| @@ -30,9 +30,10 @@ | |
| 30 30 | <<"lib/ecto_foundationdb/options.ex">>, |
| 31 31 | <<"lib/ecto_foundationdb/tenant.ex">>, |
| 32 32 | <<"lib/ecto_foundationdb/migrations_pj.ex">>, |
| 33 | - <<"lib/ecto_foundationdb/index.ex">>, |
| 33 | + <<"lib/ecto_foundationdb/index.ex">>,<<"lib/ecto_foundationdb/future.ex">>, |
| 34 34 | <<"lib/ecto_foundationdb/progressive_job.ex">>, |
| 35 35 | <<"lib/ecto_foundationdb/migration.ex">>,<<"lib/ecto_foundationdb/layer">>, |
| 36 | + <<"lib/ecto_foundationdb/layer/tx_insert.ex">>, |
| 36 37 | <<"lib/ecto_foundationdb/layer/index_inventory.ex">>, |
| 37 38 | <<"lib/ecto_foundationdb/layer/pack.ex">>, |
| 38 39 | <<"lib/ecto_foundationdb/layer/fields.ex">>, |
| @@ -59,5 +60,5 @@ | |
| 59 60 | <<"lib/ecto/adapters/foundationdb/ecto_adapter_transaction.ex">>, |
| 60 61 | <<"lib/ecto/adapters/foundationdb/ecto_adapter_queryable.ex">>, |
| 61 62 | <<"lib/ecto/adapters/foundationdb/ecto_adapter.ex">>,<<".formatter.exs">>, |
| 62 | - <<"mix.exs">>,<<"README.md">>,<<"LICENSE.md">>]}. |
| 63 | + <<"mix.exs">>,<<"README.md">>,<<"LICENSE.md">>,<<"CHANGELOG.md">>]}. |
| 63 64 | {<<"build_tools">>,[<<"mix">>]}. |
| @@ -341,6 +341,152 @@ defmodule Ecto.Adapters.FoundationDB do | |
| 341 341 | |
| 342 342 | See Queries for more information. |
| 343 343 | |
| 344 | + ## Upserts |
| 345 | + |
| 346 | + The FoundationDB Adapter supports the following upsert approaches. |
| 347 | + |
| 348 | + ### Upserts with `on_conflict` |
| 349 | + |
| 350 | + The following choices for `on_conflict` are supported, and they work in the manner |
| 351 | + described by the offical Ecto documentation. |
| 352 | + |
| 353 | + - `:raise`: The default. |
| 354 | + - `:nothing` |
| 355 | + - `:replace_all` |
| 356 | + - `{:replace_all_except, fields}` |
| 357 | + - `{:replace, fields}` |
| 358 | + |
| 359 | + ### Upserts with `conflict_target` |
| 360 | + |
| 361 | + If you provide `conflict_target: []`, the FoundationDB Adapter will make no effort to |
| 362 | + inspect the existence of objects in the database before insert. When an existing |
| 363 | + object is blindly overwritten, your indexes may become inconsistent, resulting in undefined |
| 364 | + behavior. You should only consider its use if one of the following is true: |
| 365 | + |
| 366 | + - (a) your schema has no indexes |
| 367 | + - (b) you know apriori that any indexed fields are unchanged at the time of upsert, or |
| 368 | + - (c) you know apirori that the primary key you're inserting doesn't already exist. |
| 369 | + |
| 370 | + For these use cases, this option may speed up the loading of initial data at scale. |
| 371 | + |
| 372 | + ## Pipelining |
| 373 | + |
| 374 | + Pipelining is the technique of sending multiple queries to a database at the same time, and |
| 375 | + then receiving the results at a later time. In doing so, you can often avoid waiting |
| 376 | + for multiple network round trips. EctoFDB uses a `async`/`await` syntax to |
| 377 | + implement pipelining. |
| 378 | + |
| 379 | + All the "queryable" repo functions have `async` versions: |
| 380 | + |
| 381 | + - `Repo.async_get/2` / `Repo.async_get/3` |
| 382 | + - `Repo.async_get!/2` / `Repo.async_get!/3` |
| 383 | + - `Repo.async_get_by/2` / `Repo.async_get_by/3` |
| 384 | + - `Repo.async_get_by!/2` / `Repo.async_get_by!/3` |
| 385 | + - `Repo.async_one/1` / `Repo.async_one/2` |
| 386 | + - `Repo.async_one!/1` / `Repo.async_one!/2` |
| 387 | + - `Repo.async_all/1` / `Repo.async_all/2` |
| 388 | + |
| 389 | + The results of these functions are provided to `Repo.await/1` inside a `Repo.transaction/2`. |
| 390 | + |
| 391 | + Consider this example. Our transaction below has 2 network round trips in serial. |
| 392 | + The "John" User is retrieved, and then the "James" users is retrieved. Finally, the |
| 393 | + list is constructed and the transaction ends. |
| 394 | + |
| 395 | + ```elixir |
| 396 | + result = MyApp.Repo.transaction(fn -> |
| 397 | + [ |
| 398 | + MyApp.Repo.get_by(User, name: "John"), |
| 399 | + MyApp.Repo.get_by(User, name: "James") |
| 400 | + ] |
| 401 | + end, prefix: tenant) |
| 402 | + |
| 403 | + [john, james] = result |
| 404 | + ``` |
| 405 | + |
| 406 | + With pipelining, we can issue 2 `async_get_by` queries without waiting for their result, |
| 407 | + and then only when we need the data, we `await` the result. Notice that the list of futures |
| 408 | + is constructed, and then we `await` on them. |
| 409 | + |
| 410 | + ```elixir |
| 411 | + result = MyApp.Repo.transaction(fn -> |
| 412 | + futures = [ |
| 413 | + MyApp.Repo.async_get_by(User, name: "John"), |
| 414 | + MyApp.Repo.async_get_by(User, name: "James") |
| 415 | + ] |
| 416 | + |
| 417 | + MyApp.Repo.await(futures) |
| 418 | + end, prefix: tenant) |
| 419 | + |
| 420 | + [john, james] = result |
| 421 | + ``` |
| 422 | + |
| 423 | + Conceptually, only 1 network round-trip is required, so this will be faster than |
| 424 | + waiting on each in sequence. |
| 425 | + |
| 426 | + ### Pipelining on range queries (interleaving waits) |
| 427 | + |
| 428 | + The functions `Repo.async_all/1` and `Repo.async_all/2` work a little bit different on the await. When the DB must |
| 429 | + return a list of results, if that list ends up being larger than `:erlfdb`'s configured `:target_bytes`, |
| 430 | + then the result set is split into multiple round-trips to the database. If your application awaits multiple |
| 431 | + such queries, then `:erlfdb`'s "interleaving waits" feature is used. |
| 432 | + |
| 433 | + Consider this example, where we're retrieving all Users and all Posts in the same transaction. |
| 434 | + |
| 435 | + ```elixir |
| 436 | + MyApp.Repo.transaction(fn -> |
| 437 | + futures = [ |
| 438 | + MyApp.Repo.async_all(User), |
| 439 | + MyApp.Repo.async_all(Post) |
| 440 | + ] |
| 441 | + |
| 442 | + MyApp.Repo.await(futures) |
| 443 | + end, prefix: tenant) |
| 444 | + ``` |
| 445 | + |
| 446 | + Each query individually executed would be multiple round-trips to the database as pages of results are retrieved. |
| 447 | + |
| 448 | + ```mermaid |
| 449 | + sequenceDiagram |
| 450 | + Client->>Server: get_range(User) |
| 451 | + Server-->>Client: {100 users, continuation_token} |
| 452 | + Client->>Server: get_range(User, continuation_token) |
| 453 | + Server-->>Client: {10 users, done} |
| 454 | + ``` |
| 455 | + |
| 456 | + With `:erlfdb`'s interleaving waits, we send as many get_range requests as possible with pipelining |
| 457 | + until all are exhausted. |
| 458 | + |
| 459 | + ```mermaid |
| 460 | + sequenceDiagram |
| 461 | + Client->>Server: get_range(User) |
| 462 | + Client->>Server: get_range(Post) |
| 463 | + Server-->>Client: {100 users, continuation_token} |
| 464 | + Server-->>Client: {100 posts, continuation_token} |
| 465 | + Client->>Server: get_range(User, continuation_token) |
| 466 | + Client->>Server: get_range(Post, continuation_token) |
| 467 | + Server-->>Client: {10 users, done} |
| 468 | + Server-->>Client: {100 posts, continuation_token} |
| 469 | + Client->>Server: get_range(Post, continuation_token) |
| 470 | + Server-->>Client: {2 posts, done} |
| 471 | + ``` |
| 472 | + |
| 473 | + This happens automatically when you include the list of futures to your `Repo.await/1`. |
| 474 | + |
| 475 | + ## Optimizing for throughput and latency |
| 476 | + |
| 477 | + Pipelining operations inside your transactions can offer significant speed-ups, and you should |
| 478 | + also consider parallelizing multiple transactions. Both techniques are useful in different |
| 479 | + circumstances. |
| 480 | + |
| 481 | + - Pipelining: The transaction can contain arbitrary logic and remain ACID compliant. If you make |
| 482 | + smart use of pipelining, the latency of execution of that logic is kept to a minimum. |
| 483 | + - Parallel transactions: Each transaction is internally ACID compliant, but any 2 given |
| 484 | + transactions may conflict if they're accessing the same keys. When transactions do not conflict, |
| 485 | + throughput goes up. |
| 486 | + |
| 487 | + If you're attempting to get the maximum throughput on data loading, you'll probably want to |
| 488 | + make use of both! |
| 489 | + |
| 344 490 | ## Standard Options |
| 345 491 | |
| 346 492 | * `:cluster_file` - The path to the fdb.cluster file. The default is |
| @@ -447,6 +593,7 @@ defmodule Ecto.Adapters.FoundationDB do | |
| 447 593 | @behaviour Ecto.Adapter.Transaction |
| 448 594 | |
| 449 595 | alias EctoFoundationDB.Database |
| 596 | + alias EctoFoundationDB.Future |
| 450 597 | alias EctoFoundationDB.Layer.Tx |
| 451 598 | alias EctoFoundationDB.Options |
| 452 599 | alias EctoFoundationDB.Tenant |
| @@ -498,7 +645,66 @@ defmodule Ecto.Adapters.FoundationDB do | |
| 498 645 | def transactional(db_or_tenant, fun), do: Tx.transactional_external(db_or_tenant, fun) |
| 499 646 | |
| 500 647 | @impl Ecto.Adapter |
| 501 | - defmacro __before_compile__(_env), do: :ok |
| 648 | + defmacro __before_compile__(_env) do |
| 649 | + quote do |
| 650 | + def async_get(queryable, id, opts \\ []), |
| 651 | + do: async_query(fn -> get(queryable, id, opts ++ [returning: {:future, :one}]) end) |
| 652 | + |
| 653 | + def async_get!(queryable, id, opts \\ []), |
| 654 | + do: async_query(fn -> get!(queryable, id, opts ++ [returning: {:future, :one}]) end) |
| 655 | + |
| 656 | + def async_get_by(queryable, clauses, opts \\ []), |
| 657 | + do: |
| 658 | + async_query(fn -> get_by(queryable, clauses, opts ++ [returning: {:future, :one}]) end) |
| 659 | + |
| 660 | + def async_get_by!(queryable, clauses, opts \\ []), |
| 661 | + do: |
| 662 | + async_query(fn -> get_by!(queryable, clauses, opts ++ [returning: {:future, :one}]) end) |
| 663 | + |
| 664 | + def async_one(queryable, opts \\ []), |
| 665 | + do: async_query(fn -> one(queryable, opts ++ [returning: {:future, :one}]) end) |
| 666 | + |
| 667 | + def async_one!(queryable, opts \\ []), |
| 668 | + do: async_query(fn -> one!(queryable, opts ++ [returning: {:future, :one}]) end) |
| 669 | + |
| 670 | + def async_all(queryable, opts \\ []), |
| 671 | + do: async_query(fn -> all(queryable, opts ++ [returning: {:future, :all}]) end) |
| 672 | + |
| 673 | + defp async_query(fun) do |
| 674 | + _res = fun.() |
| 675 | + |
| 676 | + case Process.delete(Future.token()) do |
| 677 | + nil -> |
| 678 | + raise "Pipelining failure" |
| 679 | + |
| 680 | + future -> |
| 681 | + future |
| 682 | + end |
| 683 | + after |
| 684 | + Process.delete(Future.token()) |
| 685 | + end |
| 686 | + |
| 687 | + def await(futures) do |
| 688 | + futures |
| 689 | + |> Future.await_all() |
| 690 | + |> Enum.map(fn future -> |
| 691 | + result = Future.result(future) |
| 692 | + if is_nil(result), do: raise("Pipelining failure") |
| 693 | + |
| 694 | + # Abuse a :noop option here to signal to the backend that we don't |
| 695 | + # actually want to run a query. Instead, we just want the result to |
| 696 | + # be transformed by Ecto's internal logic. |
| 697 | + case Future.all_or_one(future) do |
| 698 | + :all -> |
| 699 | + all(Future.schema(future), noop: result) |
| 700 | + |
| 701 | + :one -> |
| 702 | + one(Future.schema(future), noop: result) |
| 703 | + end |
| 704 | + end) |
| 705 | + end |
| 706 | + end |
| 707 | + end |
| 502 708 | |
| 503 709 | @impl Ecto.Adapter |
| 504 710 | defdelegate ensure_all_started(config, type), to: EctoAdapter |
| @@ -3,7 +3,7 @@ defmodule Ecto.Adapters.FoundationDB.EctoAdapter do | |
| 3 3 | @behaviour Ecto.Adapter |
| 4 4 | |
| 5 5 | @impl Ecto.Adapter |
| 6 | - defmacro __before_compile__(_env), do: :ok |
| 6 | + defmacro __before_compile__(_env), do: :this_is_never_called |
| 7 7 | |
| 8 8 | @impl Ecto.Adapter |
| 9 9 | def ensure_all_started(_config, type), do: Application.ensure_all_started(:erlfdb, type) |
| @@ -4,6 +4,7 @@ defmodule Ecto.Adapters.FoundationDB.EctoAdapterQueryable do | |
| 4 4 | |
| 5 5 | alias EctoFoundationDB.Exception.IncorrectTenancy |
| 6 6 | alias EctoFoundationDB.Exception.Unsupported |
| 7 | + alias EctoFoundationDB.Future |
| 7 8 | alias EctoFoundationDB.Layer.Fields |
| 8 9 | alias EctoFoundationDB.Layer.Ordering |
| 9 10 | alias EctoFoundationDB.Layer.Query |
| @@ -38,15 +39,36 @@ defmodule Ecto.Adapters.FoundationDB.EctoAdapterQueryable do | |
| 38 39 | } |
| 39 40 | }, {_limit, limit_fn}, %{}, ordering_fn}}, |
| 40 41 | params, |
| 41 | - _options |
| 42 | + options |
| 42 43 | ) do |
| 43 | - {context, query = %Ecto.Query{prefix: tenant}} = assert_tenancy!(query, adapter_opts) |
| 44 | + case options[:noop] do |
| 45 | + query_result when not is_nil(query_result) -> |
| 46 | + # This is the trick to load structs after a pipelined 'get'. See async_get_by, await, etc |
| 47 | + query_result |
| 44 48 | |
| 45 | - tenant |
| 46 | - |> execute_all(adapter_meta, context, query, params) |
| 47 | - |> ordering_fn.() |
| 48 | - |> limit_fn.() |
| 49 | - |> select(Fields.parse_select_fields(select_fields)) |
| 49 | + _ -> |
| 50 | + {context, query = %Ecto.Query{prefix: tenant}} = assert_tenancy!(query, adapter_opts) |
| 51 | + |
| 52 | + future = execute_all(tenant, adapter_meta, context, query, params) |
| 53 | + |
| 54 | + future = |
| 55 | + Future.apply(future, fn {objs, _continuation} -> |
| 56 | + objs |
| 57 | + |> ordering_fn.() |
| 58 | + |> limit_fn.() |
| 59 | + |> select(Fields.parse_select_fields(select_fields)) |
| 60 | + end) |
| 61 | + |
| 62 | + case options[:returning] do |
| 63 | + {:future, all_or_one} -> |
| 64 | + Process.put(Future.token(), Future.set_all_or_one(future, all_or_one)) |
| 65 | + {0, []} |
| 66 | + |
| 67 | + _ -> |
| 68 | + # Future: If there is a wrapping transaction without an `async_*` qualifier, the wait happens here |
| 69 | + Future.result(future) |
| 70 | + end |
| 71 | + end |
| 50 72 | end |
| 51 73 | |
| 52 74 | def execute( |
| @@ -167,8 +189,7 @@ defmodule Ecto.Adapters.FoundationDB.EctoAdapterQueryable do | |
| 167 189 | # 4. Post-get filtering (Remove :not_found, remove index conflicts, ) |
| 168 190 | # 5. Arrange fields based on the select input |
| 169 191 | plan = QueryPlan.get(source, schema, context, wheres, [], params) |
| 170 | - {objs, _continuation} = Query.all(tenant, adapter_meta, plan) |
| 171 | - objs |
| 192 | + Query.all(tenant, adapter_meta, plan) |
| 172 193 | end |
| 173 194 | |
| 174 195 | defp execute_update_all( |
| @@ -235,10 +256,16 @@ defmodule Ecto.Adapters.FoundationDB.EctoAdapterQueryable do | |
| 235 256 | {:halt, acc} |
| 236 257 | |
| 237 258 | acc = %{plan: plan, continuation: continuation} -> |
| 238 | - {objs, continuation} = |
| 239 | - Query.all(tenant, adapter_meta, plan, query_options.(continuation)) |
| 259 | + future = Query.all(tenant, adapter_meta, plan, query_options.(continuation)) |
| 240 260 | |
| 241 | - {[select(objs, field_names)], %{acc | continuation: continuation}} |
| 261 | + future = |
| 262 | + Future.apply(future, fn {objs, continuation} -> |
| 263 | + {[select(objs, field_names)], %{acc | continuation: continuation}} |
| 264 | + end) |
| 265 | + |
| 266 | + # We can't carry the future beyond this point, because we need the continuation. |
| 267 | + # It would be unusual to use stream inside a transaction anyway. |
| 268 | + Future.result(future) |
| 242 269 | end |
| 243 270 | |
| 244 271 | after_fun = fn _acc -> |
Loading more files…