Module: ActiveRecord::Materialized

Defined in:
lib/activerecord/materialized/module_api.rb,
lib/activerecord/materialized/view.rb,
lib/activerecord/materialized/tasks.rb,
lib/activerecord/materialized/railtie.rb,
lib/activerecord/materialized/version.rb,
lib/activerecord/materialized/metadata.rb,
lib/activerecord/materialized/registry.rb,
lib/activerecord/materialized/cold_read.rb,
lib/activerecord/materialized/refresher.rb,
lib/activerecord/materialized/reconciler.rb,
lib/activerecord/materialized/table_swap.rb,
lib/activerecord/materialized/read_router.rb,
lib/activerecord/materialized/refresh_job.rb,
lib/activerecord/materialized/view_loader.rb,
lib/activerecord/materialized/write_change.rb,
lib/activerecord/materialized/write_outbox.rb,
lib/activerecord/materialized/change_source.rb,
lib/activerecord/materialized/configuration.rb,
lib/activerecord/materialized/data_verifier.rb,
lib/activerecord/materialized/reconcile_job.rb,
lib/activerecord/materialized/summary_delta.rb,
lib/activerecord/materialized/refresh_result.rb,
lib/activerecord/materialized/async_refresher.rb,
lib/activerecord/materialized/instrumentation.rb,
lib/activerecord/materialized/metadata/schema.rb,
lib/activerecord/materialized/metadata_record.rb,
lib/activerecord/materialized/partition_state.rb,
lib/activerecord/materialized/schema_verifier.rb,
lib/activerecord/materialized/view_definition.rb,
lib/activerecord/materialized/delta_maintainer.rb,
lib/activerecord/materialized/partition_filter.rb,
lib/activerecord/materialized/partition_record.rb,
lib/activerecord/materialized/reconcile_result.rb,
lib/activerecord/materialized/source_watermark.rb,
lib/activerecord/materialized/debezium_envelope.rb,
lib/activerecord/materialized/maintenance_delta.rb,
lib/activerecord/materialized/maintenance_store.rb,
lib/activerecord/materialized/migration_builder.rb,
lib/activerecord/materialized/query_expressions.rb,
lib/activerecord/materialized/refresh_callbacks.rb,
lib/activerecord/materialized/refresh_scheduler.rb,
lib/activerecord/materialized/write_maintenance.rb,
lib/activerecord/materialized/aggregate_analysis.rb,
lib/activerecord/materialized/cache_table_schema.rb,
lib/activerecord/materialized/connection_routing.rb,
lib/activerecord/materialized/partition_snapshot.rb,
lib/activerecord/materialized/dependency_registry.rb,
lib/activerecord/materialized/incremental_refresh.rb,
lib/activerecord/materialized/metadata/timestamps.rb,
lib/activerecord/materialized/write_outbox_record.rb,
lib/activerecord/materialized/dependency_trackable.rb,
lib/activerecord/materialized/table_model_registry.rb,
lib/activerecord/materialized/column_type_inference.rb,
lib/activerecord/materialized/partition_keyed_store.rb,
lib/activerecord/materialized/relation_cache_writer.rb,
lib/activerecord/materialized/summary_delta_builder.rb,
lib/activerecord/materialized/incremental_maintainer.rb,
lib/activerecord/materialized/configurable_table_name.rb,
lib/activerecord/materialized/metadata/reconciliation.rb,
lib/activerecord/materialized/source_watermark_record.rb,
lib/activerecord/materialized/data_verification_result.rb,
lib/activerecord/materialized/maintenance_delta_builder.rb,
lib/activerecord/materialized/metadata/maintenance_payload.rb,
lib/activerecord/materialized/view_incremental_class_methods.rb,
lib/activerecord/materialized/view_query_access_class_methods.rb,
lib/activerecord/materialized/view_configuration_class_methods.rb,
lib/activerecord/materialized/view_refresh_policy_class_methods.rb

Overview

Application-level materialized views for Rails/ActiveRecord on databases without native materialized-view support (MySQL, MariaDB, SQLite).

Define views by subclassing View. This module is the top-level entry point for global configuration and operational helpers such as Materialized.verify_schema!.

See Also:

Defined Under Namespace

Modules: CacheTableSchema, ChangeSource, ColumnTypeInference, ConfigurableTableName, ConnectionRouting, DebeziumEnvelope, DependencyTrackable, Instrumentation, QueryExpressions, RefreshCallbacks, Tasks, ViewConfigurationClassMethods, ViewIncrementalClassMethods, ViewQueryAccessClassMethods, ViewRefreshPolicyClassMethods, WriteOutbox Classes: AggregateAnalysis, AsyncRefresher, ColdRead, Configuration, DataVerificationResult, DataVerifier, DeltaMaintainer, DependencyRegistry, IncrementalMaintainer, IncrementalRefresh, MaintenanceDelta, MaintenanceDeltaBuilder, MaintenanceStore, Metadata, MetadataRecord, MigrationBuilder, NotMaterializedError, PartitionFilter, PartitionKeyedStore, PartitionRecord, PartitionSnapshot, PartitionState, Railtie, ReadRouter, ReconcileJob, ReconcileResult, Reconciler, RefreshJob, RefreshResult, RefreshScheduler, Refresher, Registry, RelationCacheWriter, SchemaVerifier, SourceWatermark, SourceWatermarkRecord, SummaryDelta, SummaryDeltaBuilder, TableModelRegistry, TableSwap, View, ViewDefinition, ViewLoader, WriteChange, WriteMaintenance, WriteOutboxRecord

Constant Summary collapse

VERSION =

The gem version.

"0.2.0"

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.configuration ⇒ Configuration

The global configuration object. Prefer configure for setting values.

Returns:

  • (Configuration) —

    the current configuration (created on first use)



20
21
22
23
24
25
26
27
 
# File 'lib/activerecord/materialized/module_api.rb', line 20

def configuration
  config = @configuration
  if config.nil?
    config = Configuration.new
    @configuration = config
  end
  config
end

Class Method Details

.atomic_swap_refresh? ⇒ Boolean

Returns:

  • (Boolean) 


235
 
# File 'lib/activerecord/materialized/module_api.rb', line 235

def atomic_swap_refresh? = configuration.atomic_swap_refresh

.configure {|config| ... } ⇒ void

This method returns an undefined value.

Configure the gem, typically from an initializer.

Examples:

config/initializers/activerecord_materialized.rb

ActiveRecord::Materialized.configure do |config|
  config.default_refresh_strategy = :async
  config.refresh_dispatcher       = :active_job
  config.default_max_staleness    = 12.hours
end

Yield Parameters:



40
41
42
 
# File 'lib/activerecord/materialized/module_api.rb', line 40

def configure(&block)
  yield(configuration)
end

.drain_write_outbox!(limit: nil) ⇒ Integer

Relay pending WriteOutbox rows (captured by database triggers for out-of-band writes) through #ingest_change, then delete them. The operational entry point run from a poller, cron, or background job when a table has trigger/outbox capture installed. Returns the number of rows relayed so a caller can loop until drained. See docs/out-of-band-writes.md.

Parameters:

  • limit (Integer, nil) (defaults to: nil) —

    max rows to relay this pass (nil = all pending)

Returns:

  • (Integer) 


233
 
# File 'lib/activerecord/materialized/module_api.rb', line 233

def drain_write_outbox!(limit: nil) = WriteOutbox.drain!(limit: limit)

.enqueue_stale_reconciles!(mode: :checksum, sample: nil) ⇒ Array<String>

Fans the periodic reconcile backstop across an ActiveJob fleet: enqueues one ReconcileJob per stale, materialized view so many workers share the (expensive) drift verification instead of one process running it serially. Run it from a SINGLE owner — a leader, a dedicated instance, or one recurring job — never as cron on every server (see the distributed-deployment guide). Requires ActiveJob; without it, use reconcile_stale!.

Parameters:

  • mode (Symbol) (defaults to: :checksum) —

    drift-check depth passed to each job (+:row_count+/+:checksum+/+:full+)

  • sample (Numeric, nil) (defaults to: nil) —

    verify a random subset (Integer count / Float fraction)

Returns:

  • (Array<String>) —

    the view keys enqueued



114
115
116
117
118
 
# File 'lib/activerecord/materialized/module_api.rb', line 114

def enqueue_stale_reconciles!(mode: :checksum, sample: nil)
  enqueue_for_each_stale_view(:enqueue_stale_reconciles!) do |view_class|
    ReconcileJob.perform_later(view_class.view_key, mode: mode, sample: sample)
  end
end

.enqueue_stale_refreshes! ⇒ Array<String>

The fan-out form of the serial ActiveRecord::Materialized::Registry.refresh_stale!: enqueues one RefreshJob per stale, materialized view. Same single-owner rule as enqueue_stale_reconciles!. Requires ActiveJob.

Returns:

  • (Array<String>) —

    the view keys enqueued



124
125
126
127
128
 
# File 'lib/activerecord/materialized/module_api.rb', line 124

def enqueue_stale_refreshes!
  enqueue_for_each_stale_view(:enqueue_stale_refreshes!) do |view_class|
    RefreshJob.perform_later(view_class.view_key)
  end
end

.ingest_change(table:, operation:, source_ts: nil, **images) ⇒ void

This method returns an undefined value.

Ingests a change described as a normalized descriptor — the CDC-friendly entry point for a consumer that has the table, operation, and (optionally) the changed key columns or full before/after images as plain data rather than an ActiveRecord object. Builds a WriteChange and publishes it via publish_write_change!, so it drives the externally-fed (+change_source :none+) views on the table; a callback-driven view is left to its own callbacks — recover one after a bulk load with mark_dirty_for_tables!.

Supply +before+/+after+ images when the stream carries them; otherwise key_attributes scopes maintenance to the affected partition(s). With neither — or a partial :update image that cannot identify both the old and new partition — maintenance widens to a full recompute (always correct). See ActiveRecord::Materialized::WriteChange.from_descriptor.

Parameters:

  • table (String) —

    the changed table

  • operation (Symbol) —

    :create, :update, or :destroy

  • key_attributes (Hash, nil) —

    the GROUP BY key columns of the changed row

  • before (Hash, nil) —

    pre-image, for +:update+/+:destroy+

  • after (Hash, nil) —

    post-image, for +:create+/+:update+

  • source_ts (Integer, nil) (defaults to: nil) —

    optional monotonic source watermark (e.g. a Debezium source.ts_ms or a per-key Kafka offset). When given, an affected partition that already applied a strictly-newer value is skipped as provably-stale — an equal value still applies, so a coarse (second-granular) timestamp is safe (best-effort, reconcile-backed) — and the view's freshness/lag becomes observable via SourceWatermark. Ignored for callback-driven and full-recompute writes.

  • images (Hash) —

    the change images/keys — key_attributes:, before:, after: (forwarded to ActiveRecord::Materialized::WriteChange.from_descriptor, which rejects any unrecognized keyword)



198
199
200
201
202
203
204
205
206
207
 
# File 'lib/activerecord/materialized/module_api.rb', line 198

def ingest_change(table:, operation:, source_ts: nil, **images)
  unless source_ts.nil? || source_ts.is_a?(Integer)
    raise ArgumentError,
          "source_ts must be an Integer (e.g. a Debezium ts_ms or a Kafka offset); got #{source_ts.class}"
  end

  change = WriteChange.from_descriptor(table_name: table, operation: operation, **images)
  change = change.with_source_ts(source_ts) if source_ts
  publish_write_change!(change)
end

.ingest_debezium_change(envelope, table = nil) ⇒ void

This method returns an undefined value.

Relay one Debezium CDC change envelope through #ingest_change — mapping op (c/r → create, u → update, d → destroy), the +before+/+after+ row images, source.table, and the envelope's source.ts_ms as the source_ts watermark for you (a nested payload is unwrapped). A nil envelope (a Kafka tombstone) is a no-op; a non-tombstone with no op, or an unsupported op, raises. Pass table to override the target when it differs from source.table. table is positional (not a keyword) so an inline envelope literal is not misparsed as keyword arguments. See DebeziumEnvelope and the CDC section of the README.

Parameters:

  • envelope (Hash, nil) —

    a decoded Debezium change event value (string or symbol keys)

  • table (String, Symbol, nil) (defaults to: nil) —

    target-table override; defaults to source.table

Raises:

  • (ArgumentError) —

    on a mis-shaped envelope, an unsupported op, or an undeterminable table



221
222
223
224
 
# File 'lib/activerecord/materialized/module_api.rb', line 221

def ingest_debezium_change(envelope, table = nil)
  descriptor = DebeziumEnvelope.to_change_descriptor(envelope, table)
  ingest_change(**descriptor) if descriptor
end

.mark_dirty_for_tables!(tables) ⇒ void

This method returns an undefined value.

Coarse ingestion signal for callers that cannot describe the individual write: enqueues a full recompute for every view depending on any of tables and schedules it per each view's refresh strategy (inline for :immediate, in the background for :async). Idempotent, so it is safe to call repeatedly and to recover a view after a callback-skipping bulk load.

Parameters:

  • tables (Array<String>) —

    dependency table names



166
167
168
 
# File 'lib/activerecord/materialized/module_api.rb', line 166

def mark_dirty_for_tables!(tables)
  DependencyRegistry.mark_dirty_for_tables!(tables)
end

.metadata_table_name ⇒ Object 



44
 
# File 'lib/activerecord/materialized/module_api.rb', line 44

def metadata_table_name = configuration.metadata_table_name

.partition_table_name ⇒ Object 



46
 
# File 'lib/activerecord/materialized/module_api.rb', line 46

def partition_table_name = configuration.partition_table_name

.publish_write_change!(change) ⇒ void

This method returns an undefined value.

Publishes a committed dependency write from a custom change source (a CDC/replication stream, a bulk loader, another service). It drives the externally-fed views (change_source :none) that depend on the table; callback-driven views are left to their own commit callbacks so no view is maintained twice. To recover a callback-driven view after a callback-skipping bulk load, use mark_dirty_for_tables! instead.

Parameters:

  • change (WriteChange) —

    the committed write to publish



154
155
156
 
# File 'lib/activerecord/materialized/module_api.rb', line 154

def publish_write_change!(change)
  DependencyRegistry.publish_write_change!(change)
end

.reconcile!(mode: :checksum, sample: nil) ⇒ Array<ReconcileResult>

Reconciles every registered view — verifies its contents against the source and repairs any drift with scoped maintenance (never a full rebuild), returning a ReconcileResult per view. See Reconciler.

Parameters:

  • mode (Symbol) (defaults to: :checksum) —

    drift-check depth: :row_count, :checksum, or :full

  • sample (Numeric, nil) (defaults to: nil) —

    verify a random subset (Integer count / Float fraction)

Returns:



90
91
92
 
# File 'lib/activerecord/materialized/module_api.rb', line 90

def reconcile!(mode: :checksum, sample: nil)
  Registry.reconcile_all!(mode: mode, sample: sample)
end

.reconcile_stale!(mode: :checksum, sample: nil) ⇒ Array<ReconcileResult>

Like reconcile! but only for stale views — dirty, never refreshed, or past max_staleness — the scheduled bounded-staleness backstop (fresh views are skipped). Drive it from cron or ActiveJob.

Parameters:

  • mode (Symbol) (defaults to: :checksum) —

    drift-check depth: :row_count, :checksum, or :full

  • sample (Numeric, nil) (defaults to: nil) —

    verify a random subset (Integer count / Float fraction)

Returns:



101
102
103
 
# File 'lib/activerecord/materialized/module_api.rb', line 101

def reconcile_stale!(mode: :checksum, sample: nil)
  Registry.reconcile_stale!(mode: mode, sample: sample)
end

.verify_data(mode: :checksum, sample: nil) ⇒ Array<DataVerificationResult>

Checks every registered view's materialized contents against its source relation and returns a DataVerificationResult per view (never raises on drift), so callers can alert on or repair the divergent partition keys.

Parameters:

  • mode (Symbol) (defaults to: :checksum) —

    :row_count, :checksum, or :full

  • sample (Numeric, nil) (defaults to: nil) —

    verify a random subset (Integer count / Float fraction)

Returns:



66
67
68
 
# File 'lib/activerecord/materialized/module_api.rb', line 66

def verify_data(mode: :checksum, sample: nil)
  Registry.all.map { |view_class| DataVerifier.new(view_class, mode: mode, sample: sample).verify }
end

.verify_data!(mode: :checksum, sample: nil) ⇒ Array<DataVerificationResult>

Like verify_data but raises ActiveRecord::Materialized::DataVerifier::DataDriftError if any view has drifted — for a boot/CI/cron gate. Returns the results when all are clean.

Returns:

Raises:



75
76
77
78
79
80
81
 
# File 'lib/activerecord/materialized/module_api.rb', line 75

def verify_data!(mode: :checksum, sample: nil)
  results = verify_data(mode: mode, sample: sample)
  drifted = results.select(&:drifted?)
  return results if drifted.empty?

  raise DataVerifier::DataDriftError, data_drift_message(drifted)
end

.verify_schema! ⇒ void

This method returns an undefined value.

Verifies every registered view's cache table still matches the columns its source relation projects — run it at boot or in CI to catch a view whose definition changed without a migration. Never alters tables.

Raises:



54
55
56
57
 
# File 'lib/activerecord/materialized/module_api.rb', line 54

def verify_schema!
  registered = Registry.all
  registered.each { |view_class| SchemaVerifier.new(view_class).verify! }
end

.warn_if_in_process_dispatcher!(logger: default_logger) ⇒ void

This method returns an undefined value.

Logged once at boot (see Railtie) when ActiveJob is available but the effective dispatcher is still the in-process refresher — it does not coordinate across processes, so it is a correctness/efficiency hazard in a multi-process deployment. Silent when dispatching via ActiveJob, and silent when ActiveJob isn't loaded at all (a genuinely single-process app, for which the in-process refresher is the right and only choice).

Parameters:

  • logger (#warn, nil) (defaults to: default_logger) —

    destination (defaults to the Rails logger)



138
139
140
141
142
143
 
# File 'lib/activerecord/materialized/module_api.rb', line 138

def warn_if_in_process_dispatcher!(logger: default_logger)
  return unless configuration.active_job_available?
  return if configuration.refresh_dispatcher == :active_job

  logger&.warn(IN_PROCESS_DISPATCHER_WARNING)
end