Add change detection and watermark tracking to extractors

[ravisuhag]

Context

Meteor currently extracts everything on every run. For high-volume sources, this is wasteful and slow. Change detection allows extractors to emit only what changed since the last run.

Scope

• Define a watermark/checkpoint interface that extractors can implement
• Store watermarks between runs (local file, config store, or Compass state)
• Update high-volume extractors to support incremental extraction:
  • BigQuery — use INFORMATION_SCHEMA timestamps for modified tables
  • Postgres/MySQL — track schema modification timestamps
  • Kafka — track topic configuration changes
• Emit change type metadata on records: created, updated, deleted
• Fall back to full extraction when no watermark exists

Design Considerations

• Watermark storage should be pluggable (local file for dev, remote store for production)
• Full extraction should remain available as a fallback or explicit mode
• Change detection accuracy varies by source — document limitations per extractor

Key Files

Area	Location
Plugin interfaces	plugins/plugin.go
Runner orchestration	runner/runner.go
Runner config	runner/config.go
CLI run command	cmd/run.go
State store (new)	state/
BigQuery extractor	plugins/extractors/bigquery/bigquery.go

Why

Incremental extraction reduces load on sources, shrinks payloads, and enables faster refresh cycles. A graph that updates in minutes instead of hours is significantly more useful for AI agents.

References

• Meteor Roadmap — Incremental and Event-Driven

[ravisuhag]

Summary

Add checkpoint-based state management to Meteor so extractors can perform incremental extraction — only fetching metadata that changed since the last successful run.

Problem

Every Meteor run does a full extraction — all tables, all pages, all topics. This means:

• Redundant work on every scheduled run (re-extracting unchanged metadata)
• Unnecessary load on source systems (API rate limits, database query cost)
• Doesn't scale as the metadata footprint grows (10K+ tables, 100K+ pages)
• No change awareness — downstream consumers can't distinguish new vs unchanged records

Proposed Design

Core Concept

Persist a checkpoint after each successful run, and pass it to the extractor on the next run.

Before:  Recipe → Extractor (full scan) → Processor → Sink
After:   Recipe → State(load) → Extractor (since checkpoint) → Processor → Sink → State(save)

1. State Store

A new top-level state/ package that reads/writes checkpoint files to a local directory:

~/.meteor/state/
  bigquery-prod.json
  postgres-analytics.json

Each file stores:

{
  "last_run": "2026-04-18T22:00:00Z",
  "last_success": "2026-04-18T22:00:00Z",
  "record_count": 4521,
  "cursor": {
    "modified_after": "2026-04-18T22:00:00Z"
  }
}

The cursor field is opaque to the framework — each extractor decides what to store (a timestamp, a page offset, a change token, a schema hash, etc.). The state store just persists and retrieves it by recipe name.

Interface:

// state/state.go
type Store interface {
    Load(recipeName string) (Checkpoint, error)
    Save(recipeName string, checkpoint Checkpoint) error
    Delete(recipeName string) error
}

type Checkpoint struct {
    LastRun     time.Time      `json:"last_run"`
    LastSuccess time.Time      `json:"last_success"`
    RecordCount int            `json:"record_count"`
    Cursor      map[string]any `json:"cursor"`
}

Initial implementation is file-based (state/filestore.go). The interface allows future backends (database, S3, etc.) without changing extractors.

2. Extractor Interface — Opt-In

The existing Extractor interface remains unchanged. Two additions enable incremental support:

Pass previous cursor via Config:

In plugins/plugin.go, add a Cursor field to the existing Config struct:

type Config struct {
    URNScope  string
    RawConfig map[string]any
    Cursor    map[string]any  // Previous checkpoint cursor, nil on first run
}

New optional interface for extractors that support checkpointing:

// plugins/plugin.go
type Checkpointer interface {
    Checkpoint() (map[string]any, error)
}

Extractors that implement Checkpointer can return their cursor after extraction. Extractors that don't implement it work exactly as today — full scan, no state saved. This is fully backwards compatible.

3. Runner Orchestration

The change in runner.Run() (in runner/runner.go):

1. Before extraction: Load checkpoint from state store by recipe name
2. Inject cursor: Pass checkpoint.Cursor into extractor's Config.Cursor
3. Run pipeline as normal (extract → process → sink)
4. After sinks succeed: If extractor implements Checkpointer, call Checkpoint() and save to state store
5. If sinks fail: Don't update checkpoint — next run retries from same point

Critical guarantee: Checkpoint is saved only after sinks succeed. This gives at-least-once delivery. If a run fails mid-way, the next run replays from the last successful checkpoint. No records are skipped.

4. Recipe-Level Control

name: bigquery-prod
source:
  name: bigquery
  scope: my-project
  config:
    project_id: my-project
sinks:
  - name: compass
    config:
      host: https://compass.example.com

Incremental behavior is automatic when:

• The extractor implements Checkpointer
• A previous checkpoint exists for the recipe

Full re-extraction can be forced with:

meteor run recipe.yml --full       # Ignore saved checkpoint
meteor run recipe.yml --reset      # Delete saved checkpoint and run full

5. Extractor Implementation Example

BigQuery (in plugins/extractors/bigquery/bigquery.go):

func (e *Extractor) Init(ctx context.Context, config plugins.Config) error {
    // ...existing init...
    if config.Cursor != nil {
        if t, ok := config.Cursor["modified_after"].(string); ok {
            e.modifiedAfter, _ = time.Parse(time.RFC3339, t)
        }
    }
    e.runStartTime = time.Now().UTC()
    return nil
}

func (e *Extractor) Extract(ctx context.Context, emit plugins.Emit) error {
    // Filter: only tables with lastModifiedTime > e.modifiedAfter
    // On first run (e.modifiedAfter is zero), extract everything
}

func (e *Extractor) Checkpoint() (map[string]any, error) {
    return map[string]any{
        "modified_after": e.runStartTime.Format(time.RFC3339),
    }, nil
}

What different extractors would checkpoint:

Extractor	Cursor Strategy
BigQuery	modified_after timestamp (table lastModifiedTime)
Postgres	modified_after or schema hash per table
Confluence	modified_after (page version.createdAt)
Notion	last_edited_time filter on search API
Kafka	Topic partition offsets or topic count hash
dbt	Manifest file hash (skip if unchanged)
GitHub	since parameter on API endpoints

Implementation Plan

Step	Scope	Files
1. state/ package — file-based checkpoint store	New package	state/state.go, state/filestore.go
2. Add Cursor field to plugins.Config	1-line change	plugins/plugin.go
3. Define Checkpointer interface	~5 lines	plugins/plugin.go
4. Wire load/save into runner.Run()	~30 lines	runner/runner.go, runner/config.go
5. Add --full / --reset flags to meteor run	CLI change	cmd/run.go
6. Implement Checkpointer on BigQuery extractor as proof	Per-extractor	plugins/extractors/bigquery/bigquery.go
7. Roll out to other extractors incrementally	Per-extractor	Ongoing

Steps 1–5 are the framework (~200 lines of new code). Step 6 proves it works end-to-end. After that, any extractor can opt in at its own pace.

Design Decisions

Why file-based state?

Meteor is a single-binary CLI tool with no external dependencies. File-based state keeps it that way. The Store interface allows future backends (Postgres, S3, Redis) without changing extractors or the runner.

Why a top-level state/ package?

Matches Meteor's convention where cross-cutting concerns (config/, recipe/, registry/) are top-level packages. The CLI needs direct access for --reset, and future commands like meteor state list would also use it.

Why opt-in per extractor?

Not all sources support efficient change detection. Some APIs don't have modified_after filters. Forcing incremental on extractors that can't support it would either break them or require them to do full scans anyway. Opt-in via Checkpointer lets each extractor adopt incrementally.

Why save checkpoint only after sink success?

If we save the checkpoint after extraction but before sinking, a sink failure would mean records are "consumed" but never delivered. The next run would skip them. Saving after sink success guarantees at-least-once delivery — the safe default for metadata pipelines.

Why opaque cursor?

Different sources have fundamentally different change detection mechanisms (timestamps, offsets, hashes, ETags, change tokens). A generic map[string]any lets each extractor store exactly what it needs without the framework imposing a schema.

Alternatives Considered

1. Recipe-level since config — User manually passes a timestamp in recipe config. Simple but requires manual tracking, error-prone, doesn't survive across scheduled runs.

2. Sink-driven dedup — Let sinks deduplicate unchanged records. Pushes complexity downstream, doesn't reduce extraction cost, and not all sinks support dedup.

3. External state store (database) — More robust but adds a dependency. Could be a future Store implementation behind the same interface.

References

• Issue Add change detection and watermark tracking to extractors #507 — Add change detection and watermark tracking to extractors
• Airbyte incremental sync — Similar cursor-based approach
• Singer tap state — State message pattern