Sublimate.Projection (Sublimate v0.1.0)

The engine that builds and runs the infrastructure (deltas table, triggers) that keeps the destination table synchronized with its source tables, without knowing anything about the table's contents.

Sublimate.Projection owns the maintenance pipeline - the deltas table, the triggers and their timing, the cascade handling, and the merge function's wrapper. A strategy fills in what is specific to a particular projection: which tables its values are read from, the shape of the deltas table, the SQL each trigger writes, and the body of the merge that applies staged deltas to the destination.

Lifecycle

A projection moves through three phases:

Install - the engine creates the deltas table, the trigger functions and triggers, and the merge function. It calls prepare/3 once at the start to let the strategy fetch or compute what the trigger and merge SQL will need.
Maintain - as the source and its joined tables change, the installed triggers stage deltas into the deltas table, using the SQL the strategy's trigger callbacks returned.
Merge - the engine's merge function applies the staged deltas to the destination table, using the strategy's merge body.

Install happens once; maintenance is continuous, driven by database triggers; merge runs on demand.

Engine responsibilities

The engine is responsible for the infrastructure common to all projections:

the deltas table, which stages changes written by triggers and consumed by merge/2.
the trigger functions and their timing, including the distinction between a leaf table's AFTER UPDATE and BEFORE DELETE triggers and an intermediate table's AFTER INSERT OR UPDATE and BEFORE DELETE triggers.
delete-cascade handling, ensuring that cascading deletes are recorded once and only once.
the PL/pgSQL wrapper for the merge function (the strategy supplies the merge body).
ensuring that the source table has an identity column before installation.

Strategy responsibilities

The strategy supplies everything specific to a projection:

the deltas table schema
the SQL emitted for each trigger branch
the merge function body

See Sublimate.ProjectionStrategy

Chain traversal helpers

To determine which source rows are affected by a change, the engine walks the projection's join chain. Which chain to follow depends on the table from which a derived value originates.

Rather than requiring strategies to reimplement join traversal, the engine exposes its join-chain primitives as part of its public API. Strategies use these functions while generating SQL:

joins_sql_without_leaf/3, leaf_to_anchor/2, source_nearest_anchor/2, forward_joins_excluding_leaf/2, emit_joins/1, emit_forward_joins/3, split_chain_at/2, changed_table_anchor/2, and the helper functions column_name_of/1, column_ref_table/1, and same_table?/2

Managing a projection

install/3 first invokes the strategy's Sublimate.ProjectionStrategy.prepare/3 callback. It then creates the deltas table, trigger functions, triggers, and merge function.
uninstall/2 removes the installed infrastructure.
merge/2 invokes the generated merge function, which applies staged changes to the destination table.

Summary

Helpers

column_name_of(qualified_column)

The column name extracted from a qualified column.

column_ref_table(qualified_column)

The table name extracted from a qualified column.

same_table?(table_a, table_b)

Checks wheter two tables have the same name.

Lifecycle

install(projection_data, repo, postgrex_options)

First invokes the strategy's Sublimate.ProjectionStrategy.prepare/3 callback. After this: creates the deltas table, trigger functions, triggers, and merge function.

merge(projection_data, opts \\ [])

Invokes the generated merge function, which applies staged changes to the destination table.

uninstall(projection_data, opts \\ [])

Removes the installed infrastructure (see install/3).

SQL generation

changed_table_anchor(at, ref)

Build the WHERE anchor for a changed intermediate table.

emit_forward_joins(forward, link_join, ref)

Emit JOIN lines for the leaf-side joins after a changed intermediate table.

emit_joins(joins)

Emit JOIN ... ON match = to lines for a list of joins, one per line.

forward_joins_excluding_leaf(chain, leaf_table)

JOIN lines for the chain excluding the leaf (which is the query's FROM).

joins_sql_without_leaf(chain, qualified_source_table, leaf_join)

Emit the JOIN clauses for a chain, excluding its two endpoints.

leaf_to_anchor(leaf_join, ref)

The leaf join's equality, with the leaf side bound to NEW/OLD instead of its alias.

source_nearest_anchor(source_nearest, ref)

The source-nearest join binds its source-side column (to) to the trigger ref.

split_chain_at(chain, table)

Splits the chain at the given table, returning {before, at, after}: the joins preceding that table, the join for the table itself, and the joins following it.

Types

database_options()

join_spec()

join_triggers()

postgrex_options()

repo()

Helpers

column_name_of(qualified_column)

@spec column_name_of(qualified_column :: String.t()) :: String.t()

The column name extracted from a qualified column.

Examples

iex> Sublimate.Projection.column_name_of("article_categories.category_id")
"category_id"

iex> Sublimate.Projection.column_name_of("classifications.categories.id")
"id"

column_ref_table(qualified_column)

@spec column_ref_table(qualified_column :: String.t()) :: String.t()

The table name extracted from a qualified column.

Examples

iex> Sublimate.Projection.column_ref_table("article_categories.category_id")
"article_categories"

iex> Sublimate.Projection.column_ref_table("classifications.categories.id")
"classifications.categories"

same_table?(table_a, table_b)

@spec same_table?(table_a :: String.t(), table_b :: String.t()) :: boolean()

Checks wheter two tables have the same name.

Examples

iex> Sublimate.Projection.same_table?("article_categories", "classifications.categories")
false

iex> Sublimate.Projection.same_table?("classifications.categories", "classifications.categories")
true

Lifecycle

install(projection_data, repo, postgrex_options)

@spec install(Sublimate.ProjectionData.t(), repo(), postgrex_options()) ::
  {:ok, String.t()}
  | {:error, :column_not_found, String.t()}
  | {:error, Exception.t()}

First invokes the strategy's Sublimate.ProjectionStrategy.prepare/3 callback. After this: creates the deltas table, trigger functions, triggers, and merge function.

merge(projection_data, opts \\ [])

@spec merge(Sublimate.ProjectionData.t(), database_options()) ::
  :ok | {:error, [Exception.t()]}

Invokes the generated merge function, which applies staged changes to the destination table.

uninstall(projection_data, opts \\ [])

@spec uninstall(Sublimate.ProjectionData.t(), database_options()) ::
  {:ok, String.t()} | {:error, Exception.t()}

Removes the installed infrastructure (see install/3).

SQL generation

changed_table_anchor(at, ref)

@spec changed_table_anchor(at :: join_spec(), ref :: String.t()) :: String.t()

Build the WHERE anchor for a changed intermediate table.

When an intermediate (link) table changes, its own join equality becomes the anchor that ties the change back to the source rows, with the intermediate table's match-side column bound to the trigger row (NEW/OLD) instead of a joined alias.

Examples

iex> Sublimate.Projection.changed_table_anchor(%{
...>   match: "tags.id",
...>   to: "article_tags.tag_id"
...> }, "NEW")
"article_tags.tag_id = NEW.id"

emit_forward_joins(forward, link_join, ref)

@spec emit_forward_joins(
  forward :: [join_spec()],
  link_join :: join_spec(),
  ref :: String.t()
) :: String.t()

Emit JOIN lines for the leaf-side joins after a changed intermediate table.

Like emit_joins/1, but the join whose to points back at the changed table (link_join) binds that side to the trigger ref (NEW/OLD) instead of an alias, because the changed table is the trigger row, not a table in scope.

Examples

iex> forward_joins = [
...>   %{
...>     match: "classifications.categories.id",
...>     table: "classifications.categories",
...>     to: "article_categories.category_id"
...>   }
...> ]
...> Sublimate.Projection.emit_forward_joins(forward_joins,
...>   %{
...>     match: "article_categories.article_id",
...>     table: "article_categories",
...>     to: "articles.id"
...>   }, "NEW"
...> )
"JOIN classifications.categories categories ON categories.id = NEW.category_id"

iex> forward_joins = [
...>   %{
...>     match: "classifications.categories.id",
...>     table: "classifications.categories",
...>     to: "article_categories.category_id"
...>   },
...>   %{
...>     match: "classifications.category_texts.category_id",
...>     table: "classifications.category_texts",
...>     to: "classifications.categories.id"
...>   }
...> ]
...> Sublimate.Projection.emit_forward_joins(forward_joins,
...>   %{
...>     match: "article_categories.article_id",
...>     table: "article_categories",
...>     to: "articles.id"
...>   }, "NEW"
...> )
"JOIN classifications.categories categories ON categories.id = NEW.category_id
JOIN classifications.category_texts category_texts ON category_texts.category_id = categories.id"

iex> forward_joins = []
...> Sublimate.Projection.emit_forward_joins(forward_joins,
...>   %{
...>     match: "article_categories.article_id",
...>     table: "article_categories",
...>     to: "articles.id"
...>   }, "NEW"
...> )
""

emit_joins(joins)

@spec emit_joins(joins :: [join_spec()]) :: String.t()

Emit JOIN ... ON match = to lines for a list of joins, one per line.

Each join's table is qualified and aliased; an optional :where on a join is appended to its ON clause. Both sides of the equality are resolved to their aliased columns - use this for joins whose endpoints are all tables in scope, as opposed to emit_forward_joins/3, where one endpoint is the trigger row.

Examples

iex> joins = [
...>   %{
...>     match: "article_categories.article_id",
...>     table: "article_categories",
...>     to: "articles.id"
...>   }
...> ]
...> Sublimate.Projection.emit_joins(joins)
"JOIN public.article_categories article_categories ON article_categories.article_id = articles.id"

iex> joins = [
...>   %{
...>     match: "article_categories.article_id",
...>     table: "article_categories",
...>     to: "articles.id"
...>   },
...>   %{
...>     match: "classifications.categories.id",
...>     table: "classifications.categories",
...>     to: "article_categories.category_id"
...>   }
...> ]
...> Sublimate.Projection.emit_joins(joins)
"JOIN public.article_categories article_categories ON article_categories.article_id = articles.id
JOIN classifications.categories categories ON categories.id = article_categories.category_id"

forward_joins_excluding_leaf(chain, leaf_table)

@spec forward_joins_excluding_leaf(
  chain :: [join_spec()],
  leaf_table :: String.t()
) :: String.t()

JOIN lines for the chain excluding the leaf (which is the query's FROM).

Like emit_joins/1, but drops the leaf join - used when the leaf table is the FROM (as in the source-join walk), so only the intermediate joins toward it are emitted. Returns an empty string when the chain is just the leaf (a direct source-to-leaf join with no intermediates).

Examples

iex> chain = [
...>   %{table: "article_categories", match: "article_categories.article_id", to: "articles.id"},
...>   %{table: "classifications.categories", match: "classifications.categories.id", to: "article_categories.category_id"}
...> ]
...> Sublimate.Projection.forward_joins_excluding_leaf(chain, "classifications.categories")
"JOIN public.article_categories article_categories ON article_categories.article_id = articles.id"

iex> chain = [
...>   %{table: "roles", match: "roles.id", to: "authors.role_id"}
...> ]
...> Sublimate.Projection.forward_joins_excluding_leaf(chain, "roles")
""

joins_sql_without_leaf(chain, qualified_source_table, leaf_join)

@spec joins_sql_without_leaf(
  chain :: [map()],
  qualified_source_table :: String.t(),
  leaf_join :: join_spec()
) :: String.t()

Emit the JOIN clauses for a chain, excluding its two endpoints.

The source table is excluded because it is the query's FROM, and the leaf is excluded because its join becomes a WHERE anchor bound to the trigger row (NEW/OLD) rather than a joined table - see leaf_to_anchor/2. What remains are the intermediate joins between them.

Given a chain, the qualified source table, and the leaf join, returns the intermediate JOIN lines as a single string (empty when the chain has no intermediates, as in a direct source-to-leaf join).

Examples

iex> chain = [
...>   %{table: "article_tags", match: "article_tags.article_id", to: "articles.id"},
...>   %{table: "tags", match: "tags.id", to: "article_tags.tag_id"}
...> ]
...> leaf_join = %{table: "tags", match: "tags.id", to: "article_tags.tag_id"}
...> Sublimate.Projection.joins_sql_without_leaf(chain, "public.articles", leaf_join)
"JOIN public.article_tags article_tags ON article_tags.article_id = articles.id"

leaf_to_anchor(leaf_join, ref)

@spec leaf_to_anchor(leaf_join :: join_spec(), ref :: String.t()) :: String.t()

The leaf join's equality, with the leaf side bound to NEW/OLD instead of its alias.

Examples

iex> Sublimate.Projection.leaf_to_anchor(%{
...>   match: "categories.id",
...>   to: "article_categories.category_id"
...> }, "NEW")
"article_categories.category_id = NEW.id"

source_nearest_anchor(source_nearest, ref)

@spec source_nearest_anchor(source_nearest :: join_spec(), ref :: String.t()) ::
  String.t()

The source-nearest join binds its source-side column (to) to the trigger ref.

match: "article_categories.article_id", to: "articles.id" -> "article_categories.article_id = NEW.id"
match: "roles.id", to: "authors.role_id" -> "roles.id = NEW.role_id"

split_chain_at(chain, table)

@spec split_chain_at(chain :: [join_spec()], table :: String.t()) ::
  {[join_spec()], join_spec(), [join_spec()]}

Splits the chain at the given table, returning {before, at, after}: the joins preceding that table, the join for the table itself, and the joins following it.

Used to isolate an intermediate table within a chain - for example, to split at the table whose trigger fired, separating the source-side joins from the leaf-side ones.

Examples

iex> chain = [
...>   %{table: "article_tags", match: "article_tags.article_id", to: "articles.id"},
...>   %{table: "tags", match: "tags.id", to: "article_tags.tag_id"}
...> ]
...> Sublimate.Projection.split_chain_at(chain, "article_tags")
{
  [],
  %{table: "article_tags", match: "article_tags.article_id", to: "articles.id"},
  [
    %{table: "tags", match: "tags.id", to: "article_tags.tag_id"}
  ]
}

iex> chain = [
...>   %{
...>     table: "article_categories",
...>     match: "article_categories.article_id",
...>     to: "articles.id"
...>   },
...>   %{
...>     table: "classifications.categories",
...>     match: "classifications.categories.id",
...>     to: "article_categories.category_id"
...>   },
...>   %{
...>     table: "classifications.category_texts",
...>     match: "classifications.category_texts.category_id",
...>     to: "classifications.categories.id"
...>   }
...> ]
...> Sublimate.Projection.split_chain_at(chain, "classifications.categories")
{
  [
    %{
      match: "article_categories.article_id",
      table: "article_categories",
      to: "articles.id"
    }
  ],
  %{
    match: "classifications.categories.id",
    table: "classifications.categories",
    to: "article_categories.category_id"
  },
  [
    %{
      match: "classifications.category_texts.category_id",
      table: "classifications.category_texts",
      to: "classifications.categories.id"
    }
  ]
}

Types

database_options()

@type database_options() :: [
  {:repo, repo()} | {:otp_app, atom()} | postgrex_options()
]

join_spec()

@type join_spec() :: %{
  table: String.t(),
  match: String.t(),
  to: String.t(),
  where: String.t() | nil
}

join_triggers()

@type join_triggers() :: %{
  required(String.t()) => %{
    trigger: String.t(),
    trigger_fn: String.t(),
    delete_trigger: String.t()
  }
}

postgrex_options()

@type postgrex_options() :: [timeout: integer() | :infinity, log: boolean()]

repo()

@type repo() :: module()