🔄 Universal Change Data Capture (CDC) Architecture

The Veloctra Data Platform features a unified Change Data Capture framework engineered for high-throughput, low-latency delta replication. It both implements its own vectorized delta and hash-diff engines in-app and integrates directly with native database oplogs and streaming brokers without relying on heavy external JVM middleware like Debezium.

⚡ In-App & Integrated CDC 🔍 Zero-Column SHA-256 Hash Diff 📜 MongoDB Oplog Change Streams 🔄 SIMD Vector Batch Splitting

📐 In-App Implementation vs. External Integration

Veloctra addresses the classic data engineering dilemma by combining internal execution algorithms with open protocol integrations:

⚙️

Implemented In-App

Native Python / PyArrow algorithms running inside the execution pipeline:

  • ChecksumDiffCDC: Zero-timestamp row checksum hash engine
  • Dynamic high-watermark filter rewrite & batch maximum tracking
  • Atomic chunk checkpointing in StateStore (SQLite / MongoDB)
  • Zero-copy SIMD batch splitting (Deletes vs Upserts)
🔌

Integrated with External Systems

Native protocol listeners and target reconciliation engines:

  • MongoChangeStreamCDC: MongoDB replica set oplog watcher
  • Streaming message queues: Kafka, RabbitMQ, SQS, Redis Streams
  • PostgreSQL & SQLite atomic ON CONFLICT DO UPDATE upserts
  • Target SQL DELETE WHERE key = ? batch execution

🌊 Three CDC Operational Modes

🌊

1. High-Watermark Delta Sync

Designed for tables with modification timestamps (updated_at, modified_ts) or sequential IDs. Dynamically injects clauses into SQL and NoSQL extract queries and updates watermark checkpoints atomically per batch.

SQL & MongoDB Atomic StateStore Checkpoints
🔍

2. Checksum Hash-Diff CDC

Built for legacy tables with zero timestamp columns, no triggers, and no replication stream access. Maintains persistent SHA-256 hash maps of non-key attributes to detect INSERT, UPDATE, and hard DELETE events.

Zero-Column / No-Timestamp SHA-256 State Map
📜

3. MongoDB Change Streams

Subscribes to MongoDB change streams (coll.watch()) with persistent resume_after tokens, consuming continuous CRUD change events and transforming them into vectorized PyArrow record batches.

MongoDB Oplog Resume Tokens

⚡ PyArrow Vector Reconciliation & SIMD Split Pipeline

Veloctra standardizes all incoming change events into Apache Arrow RecordBatch schemas carrying unified CDC contract metadata (_cdc_op, _cdc_ts, _cdc_key). During the load phase, zero-copy SIMD filters split the stream for parallel destination routing:

Phase 1 • Extraction & Delta Detection
Source Delta Extraction (Watermark, Checksum Diff, or Change Stream)
Extracts rows based on last committed watermark (WHERE updated_at > 'last_wm'), compares full snapshot row hashes against stored state, or captures change streams from MongoDB oplogs.
▼
Phase 2 • Unified Vector Metadata Contract
Arrow Vector Batch Assembly (_cdc_op, _cdc_ts, _cdc_key)
Injects columnar operation tags (INSERT, UPDATE, DELETE), Unix epoch timestamps, and compound key strings into the memory-contiguous PyArrow batch.
▼
Phase 3 • SIMD Vector Split & Target Reconciliation
Dual-Channel Target Dispatch (Bulk Deletes vs Bulk Upserts)
pyarrow.compute.equal(_cdc_op, "DELETE") partitions the chunk into:
• Deletes: Targeted SQLConnector.bulk_delete() using match keys.
• Upserts: High-speed bulk_upsert() with ON CONFLICT (match_keys) DO UPDATE or MongoDB ReplaceOne(upsert=True).

📊 Capability & Trade-Off Matrix

CDC Mode Supported Databases Prerequisites Deletion Handling Throughput Profile
High-Watermark Delta Sync PostgreSQL, MySQL, SQLite, MongoDB, Cassandra, DynamoDB Timestamp or sequential integer column (e.g. updated_at, id) Soft deletes (e.g. is_deleted = true) ⚡ 120,000+ rows/sec
Checksum Hash-Diff CDC Any SQL / NoSQL table, CSV, Flat File without timestamps Zero modifications. Works on legacy & read-only databases Full hard deletes (missing key detection) ⚡ 50,000+ rows/sec (SHA-256 SIMD)
MongoDB Change Streams / Oplog MongoDB Replica Sets / Atlas, Redis Streams, Kafka Database change feed / oplog access Full hard & soft deletes with change event flags ⚡ Real-time continuous (< 50ms latency)

⚙️ Offset Governance & State Replay Policies

When modifying a running CDC streaming pipeline, Veloctra enforces explicit offset conflict resolution through both the Studio UI and the REST API (POST /pipelines/config):

Available Offset Update Strategies
replay_from_start: Clears stored high-watermarks and state hash tables, initiating a full historical bulk backfill before switching to delta stream synchronization.
process_new_only: Preserves existing watermark checkpoints and hash tables, capturing only new transactions created after the modification point.

🧪 Production CDC Pipeline YAML Specification

Below is a production-tested YAML definition demonstrating PostgreSQL-to-MongoDB delta streaming with high-watermark synchronization:

# ==============================================================================
# Veloctra Data Platform — CDC Pipeline Configuration
# Pipeline: PostgreSQL to MongoDB Claims Delta Stream
# ==============================================================================

pipeline_id: postgres_to_mongo_cdc
project_id: healthcare_prod_workspace
tenant_id: healthcare_prod_workspace
version: 2
description: "Streams incremental healthcare claims from PostgreSQL to MongoDB with automatic watermark tracking and zero data loss."

settings:
  chunk_size: 5000
  max_memory_percent: 75.0
  dlq_enabled: true
  circuit_breaker_enabled: true

# Source Configuration with Incremental Delta Sync
sources:
  - name: pg_healthcare_claims
    type: database
    connection_string: "enc:v1:..."
    query: "SELECT * FROM raw_claim_benef"
    chunk_size: 5000
    delta:
      watermark_column: updated_at
      watermark_type: timestamp
      initial_watermark: "2026-01-01T00:00:00"

# Vectorized Schema Mappings & Column Renaming
transformations:
  - type: rename_field
    field: desynpuf_id
    new_name: BeneficiaryId

  - type: rename_field
    field: clm_id
    new_name: ClaimId

  - type: select_columns
    columns:
      - BeneficiaryId
      - ClaimId
      - updated_at

# Destination Configuration with Idempotent Upserts
destinations:
  - name: mongo_claims_collection
    type: nosql
    db_type: mongodb
    connection_string: "enc:v1:..."
    database: healthcare_dw
    collection: claim_beneficiaries
    upsert_key: ClaimId
    batch_size: 5000