Schema Registry

DeltaForge’s schema registry tracks table schemas across time, enabling accurate event interpretation during replay and providing change detection for DDL operations.

Design Philosophy: Source-Owned Schemas

DeltaForge takes a fundamentally different approach to schema handling than many CDC tools. Rather than normalizing all database schemas into a universal type system, each source defines and owns its schema semantics.

This means:

• MySQL schemas capture MySQL semantics - column types like bigint(20) unsigned, varchar(255), and json are preserved exactly as MySQL defines them
• PostgreSQL schemas capture PostgreSQL semantics - arrays, custom types, and pg-specific attributes remain intact
• No lossy normalization - you don’t lose precision or database-specific information by forcing everything into a common format

This design avoids the maintenance burden of keeping a universal type system synchronized across all databases, and it ensures that downstream consumers receive schemas that accurately reflect the source database’s capabilities and constraints.

The SourceSchema Trait

Every CDC source implements the SourceSchema trait, which provides a common interface for fingerprinting and column access while allowing source-specific schema representations:

#![allow(unused)]
fn main() {
pub trait SourceSchema: Serialize + DeserializeOwned + Send + Sync {
    /// Source type identifier (e.g., "mysql", "postgres").
    fn source_kind(&self) -> &'static str;

    /// Content-addressable fingerprint for change detection.
    /// Two schemas with the same fingerprint are identical.
    fn fingerprint(&self) -> String;

    /// Column/field names in ordinal order.
    fn column_names(&self) -> Vec<&str>;

    /// Primary key column names.
    fn primary_key(&self) -> Vec<&str>;

    /// Human-readable description.
    fn describe(&self) -> String;
}
}

Fingerprinting

Schema fingerprints use SHA-256 hashing over JSON-serialized content to provide:

• Stability - the same schema always produces the same fingerprint
• Change detection - any structural change produces a different fingerprint
• Content-addressability - fingerprints can be used as cache keys or deduplication identifiers

#![allow(unused)]
fn main() {
pub fn compute_fingerprint<T: Serialize>(value: &T) -> String {
    let json = serde_json::to_vec(value).unwrap_or_default();
    let hash = Sha256::digest(&json);
    format!("sha256:{}", hex::encode(hash))
}
}

The fingerprint only includes structurally significant fields. For MySQL, this means columns and primary key are included, but engine and charset are excluded since they don’t affect how CDC events should be interpreted.

For PostgreSQL, the columns (including each column’s type OID and type modifier, pg_attribute.atttypmod) and the primary key are included. The type modifier was added in this release: after upgrading, each PostgreSQL table with a type-modified column (for example varchar(n), numeric(p,s) or timestamp(p)) registers one new schema version the first time its schema is loaded. Versions stored before the upgrade still load unchanged.

MySQL Schema Implementation

The MySqlTableSchema struct captures comprehensive MySQL table metadata:

#![allow(unused)]
fn main() {
pub struct MySqlTableSchema {
    /// Columns in ordinal order
    pub columns: Vec<MySqlColumn>,
    
    /// Primary key column names
    pub primary_key: Vec<String>,
    
    /// Storage engine (InnoDB, MyISAM, etc.)
    pub engine: Option<String>,
    
    /// Default charset
    pub charset: Option<String>,
    
    /// Default collation
    pub collation: Option<String>,
}

pub struct MySqlColumn {
    pub name: String,
    pub column_type: String,      // e.g., "bigint(20) unsigned"
    pub data_type: String,        // e.g., "bigint"
    pub nullable: bool,
    pub ordinal_position: u32,
    pub default_value: Option<String>,
    pub extra: Option<String>,    // e.g., "auto_increment"
    pub comment: Option<String>,
    pub char_max_length: Option<i64>,
    pub numeric_precision: Option<i64>,
    pub numeric_scale: Option<i64>,
}
}

Schema information is fetched from INFORMATION_SCHEMA at startup and cached for the pipeline’s lifetime.

Schema Registry Architecture

The schema registry serves three core functions:

1. Version tracking - maintains a history of schema versions per table
2. Change detection - compares fingerprints to detect DDL changes
3. Replay correlation - associates schemas with checkpoint positions for accurate replay

Schema Versions

Each registered schema version includes:

Sequence Numbers for Replay

The registry maintains a global monotonic sequence counter. When a schema is registered, it receives the next sequence number. Events carry this sequence number, enabling the replay engine to look up the correct schema version:

#![allow(unused)]
fn main() {
// During replay: find schema active at event's sequence
let schema = registry.get_at_sequence(tenant, db, table, event.schema_sequence);
}

This ensures events are always interpreted with the schema that was active when they were produced, even if the table structure has since changed.

Checkpoint Correlation

Schemas can be registered with an associated checkpoint, creating a correlation between schema versions and source positions:

#![allow(unused)]
fn main() {
registry.register_with_checkpoint(
    tenant, db, table, 
    &fingerprint, 
    &schema_json,
    Some(checkpoint_bytes),  // Optional: binlog position when schema was observed
).await?;
}

This correlation supports scenarios like:

• Replaying events from a specific checkpoint with the correct schema
• Determining which schema was active at a particular binlog position
• Rolling back schema state along with checkpoint rollback

Schema Loader

The MySqlSchemaLoader handles schema discovery and caching:

Pattern Expansion

Tables are specified using patterns that support wildcards:

The initial snapshot expands patterns to exactly the tables CDC captures.

Resolution at first use

A CDC start does not enumerate the configured tables or load their schemas: each table is resolved when it is first used (its first Relation message in PostgreSQL, its first rows in MySQL), from durable history or the live catalog, and registered. Only an initial snapshot expands the table patterns and loads the tables it copies. Concurrent first uses of one table share one load.

Caching

Resolved schemas are cached per source. The resident schemas are capped at 4,096 tables or about 64 MiB of serialized schema per source, whichever is reached first (a fixed limit, not configurable); beyond it the least recently used are evicted (deltaforge_source_schema_cache_evictions_total{tenant, source_id, engine}). Eviction never changes what a table resolves to: the source keeps a compact record (version, sequence, fingerprint) of every table resolved since it started or last changed lineage, and an evicted table is rebuilt from durable history as exactly that version, with no catalog query. These records are not bounded and are not part of the 64 MiB. If that version cannot be read back exactly, the source stops with a schema error rather than resolving the table again. Only an explicit reload (or a DDL of the table) resolves it again.

DDL Handling

Each DDL statement is attributed once, from its text and the event’s session database, to the objects it acts on, and that one attribution drives everything:

A DDL never reloads a schema from the live catalog and never touches a table it does not name. A qualified name (db.t) is attributed to db whatever database the session has active; the session database only resolves unqualified names.

API Endpoints

Reload Schemas

Force reload schemas from the database:

curl -X POST http://localhost:8080/pipelines/{name}/schemas/reload

This re-fetches and re-registers the schemas of the tables currently in use (cached), narrowed by the pipeline’s table patterns. It never enumerates the catalog: other tables load on their next use. Concurrent first uses of a table share one load.

List Cached Schemas

View currently cached schemas:

curl http://localhost:8080/pipelines/{name}/schemas

Limitations

• Durable by default - Schema versions are persisted via the storage backend’s Log primitive (SQLite by default) and replayed on restart, so they survive process restarts. A pure in-memory backend is also available for tests.
• No cross-pipeline sharing - Each pipeline maintains its own registry instance
• Pattern expansion at startup - New tables matching patterns require pipeline restart or reload

Best Practices

1. Use explicit table patterns in production to avoid accidentally capturing unwanted tables
2. Monitor schema reload times - slow reloads may indicate overly broad patterns
3. Trigger schema reload after DDL if your deployment process modifies schemas
4. Include schema version in downstream events for consumers that need schema evolution awareness

Troubleshooting

Unknown table_id Errors

WARN write_rows for unknown table_id, table_id=42

The binlog contains row events for a table not in the table_map. This happens when:

• A table was created after the CDC stream started
• Table patterns don’t match the table

Solution: Trigger a schema reload via the REST API.

Schema Fetch Returned 0 Columns

WARN schema fetch returned 0 columns, db=shop, table=orders

Usually indicates:

• Table doesn’t exist
• MySQL user lacks SELECT privilege on INFORMATION_SCHEMA
• Table was dropped between detection and schema load

Slow Schema Loading

WARN slow schema fetch, db=shop, table=orders, ms=350

Consider:

• Narrowing table patterns to reduce the number of tables
• Using exact table names instead of wildcards
• Verifying network latency to the MySQL server

Confluent Schema Registry Integration

DeltaForge supports the Confluent Schema Registry as the external schema store for Avro encoding. This is separate from DeltaForge’s internal schema registry (described above) which tracks source table schemas.

How it works

When a sink is configured with encoding: { type: avro, schema_registry_url: "..." }:

1. Schema derivation: DeltaForge derives an Avro schema from the first event’s JSON structure
2. Registration: The schema is registered with the Confluent Schema Registry (idempotent — same schema returns the same ID)
3. Caching: The schema ID is cached per subject, so subsequent events skip the HTTP call
4. Wire format: Events are encoded as [0x00][schema_id:4][avro_binary] (Confluent wire format)

Relationship to internal schema registry

The two registries serve complementary purposes — the internal registry tracks what the source looks like, while the Confluent registry stores what consumers need to decode the wire format.

Configuration

See the Envelopes and Encodings page for full Avro encoding configuration, including subject naming strategies and authentication.