Collect (Collect v0.1.0)

Copy Markdown

Collect maintains a document table: a continuously updated table that stores data from one or more source tables (optionally combined with joined tables).

Each document table row contains the source identity, a structured data jsonb and a weighted search_vector for full-text search, plus optional extracted typed columns for sorting and indexing.

It is a consumer of Sublimate ➚. While Sublimate installs the triggers and the deltas/merge pipeline, Collect defines what a document contains: which data fields, search fields, custom columns and indexes, and applying optional scope filters to conditionally include rows.

The document configuration map is the central instruction set for creating the document table. It is used to create the document table and populate it from a source table, optionally including joined tables.

Summary

Document table

create_document_table(document_table, config_or_configs, opts \\ [])

Creates the document table, installs update triggers and their trigger functions, and backfills it from the current source rows.

create_document_table_if_not_exists(document_table, config_or_configs, opts \\ [])

Identical to create_document_table/3 - will silently succeeds if the table already exists.

drop_document_table(document_table, opts \\ [])

Removes the document table and all the infrastructure installed alongside it: the deltas table, the update triggers and their trigger functions, and the merge function.

drop_document_table_from_config(document_table, config_or_configs, opts \\ [])

Identical to drop_document_table/2, but instead of discovering, this function uses the config to read triggers and trigger functions.

merge_deltas(document_table, config_or_configs, opts \\ [])

Applies staged changes to the document table, rebuilding each dirty identity's row (upserting those that qualify, deleting those that no longer do).

Helpers

table_exists?(qualified_table_name, opts \\ [])

Tests whether a table exists. Pass a qualified table name (including schema prefix) if the table is not in the "public" schema.

Types

column()
create_document_table_return()
data_field()
database_options()
document_table_config()
document_table_config_or_configs()
index()
join_spec()
join_triggers()
object_array_column()
postgrex_options()
repo()
scope()
search_field()

Document table

create_document_table(document_table, config_or_configs, opts \\ [])

@spec create_document_table(
  String.t(),
  document_table_config_or_configs(),
  database_options()
) :: create_document_table_return()

Creates the document table, installs update triggers and their trigger functions, and backfills it from the current source rows.

Returns an error if the document table already exists. See also create_document_table_if_not_exists/3 to ignore the error.

Options

  • repo (module) - the Ecto repo to run against. Defaults to the configured repo.
  • timeout (integer or :infinity) - Postgrex timeout.

Examples

config = %{
  source_table: "articles",
  add_identity_column_if_not_exists: true,
  identity_column: "identity",
  language: "english",
  data_fields: [
    %{field_name: "title", value_column: "title"},
    %{field_name: "summary", value_column: "summary"}
  ],
  search_fields: [
    %{field_name: "title", weight: 1},
    %{field_name: "summary"}
  ]
}

Collect.create_document_table("collections", config, opts)

With a globally configured repo:

Collect.create_document_table("collections", config)

In a non-public schema:

Collect.create_document_table("repository.collections", config, opts)

create_document_table_if_not_exists(document_table, config_or_configs, opts \\ [])

@spec create_document_table_if_not_exists(
  String.t(),
  document_table_config_or_configs(),
  database_options()
) :: create_document_table_return()

Identical to create_document_table/3 - will silently succeeds if the table already exists.

Options

  • repo (module) - the Ecto repo to run against. Defaults to the configured repo.
  • timeout (integer or :infinity) - Postgrex timeout.

Examples

config = %{...}
Collect.create_document_table_if_not_exists("collections", config, opts)

With a globally configured repo:

Collect.create_document_table_if_not_exists("collections", config)

In a non-public schema:

Collect.create_document_table_if_not_exists(
  "repository.collections",
  config,
  opts
)

drop_document_table(document_table, opts \\ [])

@spec drop_document_table(
  String.t(),
  database_options()
) :: {:ok, String.t()} | {:error, :table_not_found} | {:error, Exception.t()}

Removes the document table and all the infrastructure installed alongside it: the deltas table, the update triggers and their trigger functions, and the merge function.

Discovers triggers and trigger functions from the catalog. Alternatively, use drop_document_table_from_config/3 in restricted environments.

Options

  • repo (module) - the Ecto repo to run against. Defaults to the configured repo.
  • timeout (integer or :infinity) - Postgrex timeout.

Examples

drop_document_table("collections", opts)

With a globally configured repo:

drop_document_table("collections")

In a non-public schema:

drop_document_table("repository.collections", opts)

drop_document_table_from_config(document_table, config_or_configs, opts \\ [])

@spec drop_document_table_from_config(
  String.t(),
  document_table_config_or_configs(),
  database_options()
) :: {:ok, String.t()} | {:error, :table_not_found} | {:error, Exception.t()}

Identical to drop_document_table/2, but instead of discovering, this function uses the config to read triggers and trigger functions.

Use this variant in environments where reading function source / catalog introspection is restricted. Some locked-down database environments or permission setups might restrict reading function source (prosrc), which prohibits the discovery method from drop_document_table.

Options

  • repo (module) - the Ecto repo to run against. Defaults to the configured repo.
  • timeout (integer or :infinity) - Postgrex timeout.

Examples

config = %{...}
drop_document_table_from_config("collections", config, opts)

With a globally configured repo:

drop_document_table_from_config("collections", config)

In a non-public schema:

drop_document_table_from_config("repository.collections", config, opts)

merge_deltas(document_table, config_or_configs, opts \\ [])

@spec merge_deltas(
  String.t(),
  document_table_config_or_configs(),
  database_options()
) :: :ok | {:error, term()}

Applies staged changes to the document table, rebuilding each dirty identity's row (upserting those that qualify, deleting those that no longer do).

Options

  • repo (module) - the Ecto repo to run against. Defaults to the configured repo.
  • timeout (integer or :infinity) - Postgrex timeout.

Examples

config = %{...}
Collect.merge_deltas("collections", config, opts)

With a globally configured repo:

Collect.merge_deltas("collections", config)

In a non-public schema:

Collect.merge_deltas("repository.collections", config, opts)

Helpers

table_exists?(qualified_table_name, opts \\ [])

@spec table_exists?(String.t(), database_options()) :: boolean()

Tests whether a table exists. Pass a qualified table name (including schema prefix) if the table is not in the "public" schema.

Performs a database query using pg_class.

Options

  • repo (module) - the Ecto repo to run against. Defaults to the configured repo.
  • timeout (integer or :infinity) - Postgrex timeout.

Examples

Collect.table_exists?("collections", opts)

With a globally configured repo:

Collect.table_exists?("collections")

In a non-public schema:

Collect.table_exists?("repository.collections", opts)

Types

column()

@type column() :: %{
  from: String.t(),
  key: String.t() | nil,
  name: String.t(),
  transforms: [String.t()] | nil,
  type: String.t()
}

create_document_table_return()

@type create_document_table_return() ::
  {:ok, String.t()}
  | {:error, :column_not_found, String.t()}
  | {:error, :column_type_conflict, {String.t(), term(), term()}}
  | {:error, :configs_target_different_destinations, [String.t()]}
  | {:error, :document_table_exists, String.t()}
  | {:error, :duplicate_source_names, [String.t()]}
  | {:error, :identity_column_already_exists, String.t()}
  | {:error, :identity_column_invalid_type, String.t()}
  | {:error, :identity_column_not_found, String.t()}
  | {:error, :invalid_table_name, String.t()}
  | {:error, :no_configs}
  | {:error, :table_names_not_distinct}
  | {:error, Exception.t()}

data_field()

@type data_field() :: %{
  aggregate: :array | :first | :object_array | nil,
  columns: [object_array_column()] | nil,
  field_name: String.t(),
  join_table: String.t() | nil,
  joins: [join_spec()] | nil,
  value_column: String.t() | nil
}

database_options()

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

document_table_config()

@type document_table_config() ::
  %{
    add_identity_column_if_not_exists: boolean() | nil,
    columns: [column()] | nil,
    data_fields: [data_field()] | nil,
    identity_column: String.t(),
    indexes: [index()] | nil,
    language: String.t() | nil,
    scopes: [scope()] | nil,
    search_fields: [search_field()] | nil,
    source_name: String.t() | nil,
    source_table: String.t()
  }
  | keyword()

document_table_config_or_configs()

@type document_table_config_or_configs() ::
  document_table_config() | [document_table_config()]

index()

@type index() :: %{columns: [String.t()], using: String.t() | nil}

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()) => %{
    delete_trigger: String.t(),
    trigger_fn: String.t(),
    trigger: String.t()
  }
}

object_array_column()

@type object_array_column() :: %{column: String.t(), key: String.t()}

postgrex_options()

@type postgrex_options() :: [{:timeout, integer() | :infinity}]

repo()

@type repo() :: module()

scope()

@type scope() :: %{
  join_table: String.t() | nil,
  joins: [join_spec()] | nil,
  where: String.t()
}

search_field()

@type search_field() :: %{
  field_name: String.t(),
  key: String.t() | nil,
  transforms: [String.t()] | nil,
  weight: 1 | 2 | 3 | 4 | nil
}