Class: ActiveRecord::Materialized::SourceWatermark Private

Inherits:
PartitionKeyedStore show all
Defined in:
lib/activerecord/materialized/source_watermark.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Per-partition CDC source watermarks: the max applied source_ts for each (view, partition). Two jobs, both opt-in via a source_ts on ingest_change:

  • Suppression — a redelivered or out-of-order change whose watermark is < the partition's applied watermark is provably redundant (the cache already reflects a strictly-newer state), so its recompute is skipped. A change at the same watermark still applies: a coarse (e.g. second-granular) source_ts such as Debezium's MySQL source.ts_ms can tie two distinct commits, so an equal value is not treated as a redelivery. Best-effort and reconcile-backed: the watermark advances at ingest, so a crash before the recompute can over-suppress a partition until reconciliation (#64) heals it — an optimization over always-recompute, never a correctness substitute.
  • Observability — #oldest reports the view's most-behind partition watermark, so a caller can compute freshness/lag against its own source clock (in the same unit as source_ts).

Only the scoped-recompute path (a change_source :none CDC-fed view) carries a watermark; a full recompute (no partition key) is never suppressed. source_ts must be monotonic (non-decreasing) per partition (e.g. a Debezium source.ts_ms or a per-key Kafka offset); ties are safe (only a strictly-older value is suppressed). Mirrors PartitionState's per-partition store.

Instance Method Summary collapse

Methods inherited from PartitionKeyedStore

#initialize

Constructor Details

This class inherits a constructor from ActiveRecord::Materialized::PartitionKeyedStore

Instance Method Details

#oldest ⇒ Integer?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The oldest applied watermark across the view's partitions that have received a watermarked change (its most-behind such partition), or nil when none have. Subtract from the source clock (same unit as source_ts) for lag. A partition maintained only via widen-to-full changes records no watermark and so is not reflected here; a stalled stream (no events arriving) is a pipeline-level concern, monitored at the consumer. A read only — never provisions the table.

Returns:

  • (Integer, nil) 


51
52
53
54
55
 
# File 'lib/activerecord/materialized/source_watermark.rb', line 51

def oldest
  return nil unless record_class.table_exists?

  scope.minimum(:source_ts)
end

#suppress(source_ts, delta) ⇒ MaintenanceDelta?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Drop the delta's partitions that already applied a strictly-newer source_ts, and advance the watermark for the survivors — a change at the same source_ts still applies, so a distinct write sharing a coarse (e.g. second-granular) timestamp is never suppressed. A full-partition delta (no key) is returned unchanged — it can't be scoped.

Returns:

  • (MaintenanceDelta, nil) —

    the surviving partitions, or nil when every one was suppressed



32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/activerecord/materialized/source_watermark.rb', line 32

def suppress(source_ts, delta)
  return delta if delta.full_partition?

  ensure_table!
  current = current_watermarks(delta.key_tuples) # one read for the whole decision
  fresh = delta.key_tuples.reject { |tuple| (ts = current[serialize(tuple)]) && ts > source_ts }
  return nil if fresh.empty?

  advance!(fresh, source_ts)
  MaintenanceDelta.scoped(fresh)
end