Change Data Capture with Event Streaming

Executive Summary

Feature: WAL-Based CDC with Kafka/Kinesis Integration, Schema Evolution, and At-Least-Once Delivery

Target Industries: Financial services, E-commerce, Healthcare, IoT, Real-time analytics

Problem Being Solved

Current Pain Points

• High Latency: Batch ETL introduces hours of delay
• Schema Brittleness: Changes break downstream consumers
• Delivery Gaps: At-most-once causes data loss
• Operational Complexity: Separate CDC tools to manage
• Limited Sinks: Support for only specific targets
• No Filtering: All changes captured, wasting bandwidth

Why This Problem Exists

Traditional databases weren't designed for streaming:

• WAL designed for recovery, not replication
• No standard CDC format or API
• External tools add complexity and latency
• Schema changes require manual coordination
• No built-in exactly-once semantics

Technical Implementation

CDC Architecture

Transaction Processing              CDC Engine                    Streaming Targets
+-------------------+            +------------------+            +------------------+
|                   |            |                  |            |                  |
| INSERT/UPDATE/DEL |---WAL----> | WAL Parser       |            | Apache Kafka     |
| Transaction       |            | - Decode changes |            | AWS Kinesis      |
| Commit            |            | - Apply filters  |            | Azure EventHub   |
|                   |            | - Schema map     |            | Google Pub/Sub   |
+-------------------+            +------------------+            | Apache Pulsar    |
                                          |                      +------------------+
                                          v                              ^
                                 +------------------+                    |
                                 | Event Formatter  |                    |
                                 | - Avro/JSON      |----Publish---------|
                                 | - Protobuf       |
                                 | - Debezium fmt   |
                                 +------------------+
                                          |
                                          v
                                 +------------------+
                                 | Delivery Manager |
                                 | - Offset tracking|
                                 | - Exactly-once   |
                                 | - Retry logic    |
                                 +------------------+

CDC Configuration

-- Enable CDC for a table
SELECT helios_cdc_enable(
    table_name => 'orders',
    config => '{
        "capture_mode": "all_changes",
        "include_columns": ["order_id", "customer_id", "total", "status", "updated_at"],
        "exclude_columns": ["internal_notes"],
        "filter": "status != ''draft''",
        "key_columns": ["order_id"],
        "before_image": true
    }'::jsonb
);

-- Configure streaming destination
SELECT helios_cdc_add_sink(
    sink_name => 'kafka_orders',
    sink_type => 'kafka',
    config => '{
        "bootstrap_servers": "kafka-cluster:9092",
        "topic": "orders-cdc",
        "format": "avro",
        "schema_registry": "http://schema-registry:8081",
        "acks": "all",
        "compression": "snappy",
        "batch_size": 16384,
        "linger_ms": 5
    }'::jsonb
);

-- System tables for CDC monitoring
CREATE TABLE _helios_cdc_publications (
    publication_id UUID PRIMARY KEY,
    table_name TEXT,
    config JSONB,
    created_at TIMESTAMP,
    status VARCHAR(20)
);

CREATE TABLE _helios_cdc_offsets (
    sink_name TEXT,
    table_name TEXT,
    last_lsn BIGINT,
    last_timestamp TIMESTAMP,
    events_published BIGINT,
    PRIMARY KEY (sink_name, table_name)
);

Concrete Business Examples

Example 1: Real-Time Order Processing Pipeline

Scenario: E-commerce platform streaming order events to 5 downstream systems.

-- Enable CDC on orders table with full configuration
SELECT helios_cdc_enable(
    table_name => 'orders',
    config => '{
        "capture_mode": "all_changes",
        "operations": ["INSERT", "UPDATE", "DELETE"],
        "include_before_image": true,
        "include_transaction_info": true,
        "partition_key": "customer_region"
    }'::jsonb
);

-- Add multiple sinks for different consumers
SELECT helios_cdc_add_sink('kafka_fulfillment', 'kafka', '{
    "bootstrap_servers": "kafka:9092",
    "topic": "orders.fulfillment",
    "format": "avro"
}'::jsonb);

SELECT helios_cdc_add_sink('kafka_analytics', 'kafka', '{
    "bootstrap_servers": "kafka:9092",
    "topic": "orders.analytics",
    "format": "json"
}'::jsonb);

SELECT helios_cdc_add_sink('kinesis_fraud', 'kinesis', '{
    "stream_name": "orders-fraud-detection",
    "region": "us-east-1",
    "format": "json"
}'::jsonb);

-- Sample CDC event produced
/*
{
    "metadata": {
        "source": "heliosdb",
        "database": "ecommerce",
        "table": "orders",
        "operation": "UPDATE",
        "timestamp": "2024-03-15T14:23:45.123Z",
        "lsn": 12847293847,
        "transaction_id": "txn_abc123"
    },
    "before": {
        "order_id": "ORD-12345",
        "customer_id": "CUST-789",
        "total": 150.00,
        "status": "pending",
        "customer_region": "us-east"
    },
    "after": {
        "order_id": "ORD-12345",
        "customer_id": "CUST-789",
        "total": 150.00,
        "status": "confirmed",
        "customer_region": "us-east"
    },
    "diff": {
        "status": {"from": "pending", "to": "confirmed"}
    }
}
*/

-- Monitor CDC performance
SELECT * FROM helios_cdc_metrics('orders');

/*
{
    "table": "orders",
    "source_metrics": {
        "transactions_processed": 847293,
        "events_captured": 1234567,
        "capture_latency_p50_ms": 2,
        "capture_latency_p99_ms": 15,
        "wal_lag_bytes": 1234
    },
    "sink_metrics": {
        "kafka_fulfillment": {
            "events_published": 1234567,
            "publish_latency_p50_ms": 5,
            "publish_latency_p99_ms": 23,
            "lag_events": 0
        },
        "kafka_analytics": {
            "events_published": 1234567,
            "publish_latency_p50_ms": 4,
            "publish_latency_p99_ms": 18,
            "lag_events": 0
        },
        "kinesis_fraud": {
            "events_published": 1234567,
            "publish_latency_p50_ms": 12,
            "publish_latency_p99_ms": 45,
            "lag_events": 5
        }
    },
    "end_to_end_latency": {
        "p50_ms": 8,
        "p99_ms": 35,
        "p999_ms": 89
    }
}
*/

-- Query CDC stream status
SELECT
    sink_name,
    events_published,
    last_timestamp,
    NOW() - last_timestamp as lag
FROM _helios_cdc_offsets
WHERE table_name = 'orders';

/*
sink_name         | events_published | last_timestamp          | lag
------------------|------------------|-------------------------|--------
kafka_fulfillment | 123,456,789      | 2024-03-15 14:23:45.123 | 0.002s
kafka_analytics   | 123,456,789      | 2024-03-15 14:23:45.120 | 0.005s
kinesis_fraud     | 123,456,784      | 2024-03-15 14:23:45.100 | 0.025s
*/

Example 2: Financial Transaction Audit Stream

Scenario: Bank streaming all transactions for regulatory compliance and fraud detection.

-- Enable CDC with compliance features
SELECT helios_cdc_enable(
    table_name => 'transactions',
    config => '{
        "capture_mode": "full_audit",
        "operations": ["INSERT", "UPDATE", "DELETE"],
        "include_before_image": true,
        "include_user_context": true,
        "include_query_text": false,
        "compliance": {
            "redact_pii": true,
            "pii_columns": ["ssn", "account_number"],
            "redaction_method": "tokenize",
            "audit_retention_days": 2555
        },
        "exactly_once": {
            "enabled": true,
            "checkpoint_interval_ms": 1000
        }
    }'::jsonb
);

-- Stream to compliance and fraud systems
SELECT helios_cdc_add_sink('kafka_compliance', 'kafka', '{
    "topic": "transactions.audit",
    "format": "avro",
    "exactly_once": true,
    "transactional_id": "helios-cdc-compliance"
}'::jsonb);

SELECT helios_cdc_add_sink('kinesis_fraud_ml', 'kinesis', '{
    "stream_name": "transactions-fraud-ml",
    "format": "json",
    "partition_key": "account_id"
}'::jsonb);

-- Sample compliant CDC event
/*
{
    "metadata": {
        "source": "heliosdb",
        "table": "transactions",
        "operation": "INSERT",
        "timestamp": "2024-03-15T14:23:45.123Z",
        "user": "teller_system",
        "session_id": "sess_abc123",
        "client_ip": "10.0.1.100"
    },
    "after": {
        "transaction_id": "TXN-847293",
        "account_number": "tok_1234abcd",  // Tokenized
        "amount": 15000.00,
        "transaction_type": "wire_transfer",
        "beneficiary_account": "tok_5678efgh",  // Tokenized
        "timestamp": "2024-03-15T14:23:45.000Z"
    },
    "compliance": {
        "audit_id": "AUD-123456",
        "retention_until": "2031-03-15",
        "pii_tokenized": ["account_number", "beneficiary_account"]
    }
}
*/

-- Exactly-once delivery verification
SELECT * FROM helios_cdc_delivery_status('kafka_compliance');

/*
{
    "sink": "kafka_compliance",
    "delivery_mode": "exactly_once",
    "transaction_status": {
        "open_transactions": 0,
        "committed_transactions": 847293,
        "aborted_transactions": 0
    },
    "offset_tracking": {
        "current_offset": 847293847,
        "committed_offset": 847293847,
        "pending_offset": 0
    },
    "deduplication": {
        "duplicate_events_detected": 12,
        "duplicate_events_filtered": 12,
        "unique_events_delivered": 847293
    },
    "compliance_metrics": {
        "pii_redactions": 1694586,
        "audit_records_created": 847293,
        "compliance_status": "COMPLIANT"
    }
}
*/

Example 3: Schema Evolution Without Breaking Consumers

Scenario: Evolving order schema without disrupting downstream systems.

-- Original schema
CREATE TABLE orders (
    order_id VARCHAR(50) PRIMARY KEY,
    customer_id VARCHAR(50),
    total DECIMAL(10,2),
    status VARCHAR(20)
);

-- Enable CDC with schema evolution support
SELECT helios_cdc_enable('orders', '{
    "schema_evolution": {
        "enabled": true,
        "compatibility_mode": "full",
        "schema_registry": "http://schema-registry:8081",
        "auto_register": true
    }
}'::jsonb);

-- Add new column (backward compatible)
ALTER TABLE orders ADD COLUMN shipping_method VARCHAR(50) DEFAULT 'standard';

-- CDC automatically handles schema evolution
SELECT * FROM helios_cdc_schema_history('orders');

/*
{
    "table": "orders",
    "schema_versions": [
        {
            "version": 1,
            "timestamp": "2024-01-01T00:00:00Z",
            "schema_id": 1001,
            "fields": ["order_id", "customer_id", "total", "status"]
        },
        {
            "version": 2,
            "timestamp": "2024-03-15T14:00:00Z",
            "schema_id": 1002,
            "fields": ["order_id", "customer_id", "total", "status", "shipping_method"],
            "changes": [
                {"type": "ADD_FIELD", "field": "shipping_method", "default": "standard"}
            ],
            "compatibility": "BACKWARD"
        }
    ],
    "consumers": {
        "fulfillment_service": {"using_version": 2, "status": "compatible"},
        "analytics_pipeline": {"using_version": 1, "status": "compatible"},
        "legacy_system": {"using_version": 1, "status": "compatible"}
    }
}
*/

-- Breaking change protection
ALTER TABLE orders ADD COLUMN required_field VARCHAR(50) NOT NULL;

-- CDC warns about incompatible change
/*
WARNING: Schema change is not backward compatible
- Field 'required_field' is required but has no default
- Existing consumers using schema v2 will fail

Options:
1. Add default value: ALTER TABLE orders ADD COLUMN required_field VARCHAR(50) DEFAULT 'unknown'
2. Force break: SELECT helios_cdc_force_schema_break('orders')
3. Create new topic: SELECT helios_cdc_create_version_topic('orders', 3)
*/

-- Schema compatibility check
SELECT * FROM helios_cdc_check_compatibility(
    table_name => 'orders',
    proposed_change => 'ADD COLUMN priority VARCHAR(10) NOT NULL'
);

/*
{
    "compatibility_check": "FAILED",
    "proposed_change": "ADD COLUMN priority VARCHAR(10) NOT NULL",
    "issues": [
        {
            "issue": "Required field without default",
            "affected_consumers": ["fulfillment_service", "analytics_pipeline", "legacy_system"],
            "fix": "Add DEFAULT clause or make nullable"
        }
    ],
    "recommendation": "ALTER TABLE orders ADD COLUMN priority VARCHAR(10) DEFAULT 'normal'"
}
*/

Example 4: Filtered CDC with Transformation

Scenario: IoT platform streaming only critical sensor events with enrichment.

-- Enable CDC with filtering and transformation
SELECT helios_cdc_enable(
    table_name => 'sensor_readings',
    config => '{
        "capture_mode": "filtered",
        "filter_expression": "reading_value > threshold OR alert_level = ''critical''",
        "transformations": [
            {
                "type": "enrich",
                "source_column": "sensor_id",
                "lookup_table": "sensors",
                "lookup_column": "sensor_id",
                "add_columns": ["location", "sensor_type", "owner"]
            },
            {
                "type": "compute",
                "name": "deviation_pct",
                "expression": "ABS(reading_value - baseline) / baseline * 100"
            },
            {
                "type": "mask",
                "column": "internal_calibration",
                "method": "remove"
            }
        ],
        "partition_key": "sensor_type"
    }'::jsonb
);

-- Stream to analytics with transformations applied
SELECT helios_cdc_add_sink('kafka_iot_alerts', 'kafka', '{
    "topic": "iot.critical-readings",
    "format": "json",
    "include_transformations": true
}'::jsonb);

-- Sample transformed CDC event
/*
{
    "metadata": {
        "source": "heliosdb",
        "table": "sensor_readings",
        "operation": "INSERT",
        "timestamp": "2024-03-15T14:23:45.123Z",
        "filter_matched": "reading_value > threshold"
    },
    "after": {
        "reading_id": "READ-847293",
        "sensor_id": "SENS-001",
        "reading_value": 95.5,
        "threshold": 80.0,
        "alert_level": "warning",
        "timestamp": "2024-03-15T14:23:45.000Z",

        // Enriched fields from sensors table
        "location": "Building A, Floor 3",
        "sensor_type": "temperature",
        "owner": "Facilities Team",

        // Computed field
        "deviation_pct": 19.375

        // internal_calibration removed by mask
    }
}
*/

-- Monitor filtering effectiveness
SELECT * FROM helios_cdc_filter_metrics('sensor_readings');

/*
{
    "table": "sensor_readings",
    "filtering": {
        "total_changes": 12847293,
        "filter_matched": 234567,
        "filter_excluded": 12612726,
        "filter_ratio": 0.018,
        "bandwidth_saved_gb": 234.5
    },
    "transformations": {
        "enrichments_applied": 234567,
        "enrichment_cache_hit_rate": 0.95,
        "computations_performed": 234567,
        "masks_applied": 234567
    }
}
*/

Example 5: Multi-Cloud CDC Federation

Scenario: Global enterprise streaming changes to AWS, Azure, and GCP simultaneously.

-- Enable multi-cloud CDC
SELECT helios_cdc_enable_multicloud(
    table_name => 'global_customers',
    config => '{
        "regional_routing": {
            "enabled": true,
            "routing_column": "customer_region",
            "routes": {
                "americas": ["aws_kinesis", "azure_eventhub"],
                "europe": ["azure_eventhub", "gcp_pubsub"],
                "asia": ["gcp_pubsub", "aws_kinesis"]
            }
        }
    }'::jsonb
);

-- Configure multi-cloud sinks
SELECT helios_cdc_add_sink('aws_kinesis', 'kinesis', '{
    "stream_name": "customers-americas",
    "region": "us-east-1"
}'::jsonb);

SELECT helios_cdc_add_sink('azure_eventhub', 'eventhub', '{
    "namespace": "helios-cdc",
    "eventhub": "customers-europe",
    "connection_string": "..."
}'::jsonb);

SELECT helios_cdc_add_sink('gcp_pubsub', 'pubsub', '{
    "project": "helios-analytics",
    "topic": "customers-asia"
}'::jsonb);

-- Monitor multi-cloud delivery
SELECT * FROM helios_cdc_multicloud_status();

/*
{
    "table": "global_customers",
    "regional_distribution": {
        "americas": {
            "event_count": 4234567,
            "sinks": {
                "aws_kinesis": {"status": "healthy", "lag_ms": 12},
                "azure_eventhub": {"status": "healthy", "lag_ms": 45}
            }
        },
        "europe": {
            "event_count": 3567890,
            "sinks": {
                "azure_eventhub": {"status": "healthy", "lag_ms": 23},
                "gcp_pubsub": {"status": "healthy", "lag_ms": 34}
            }
        },
        "asia": {
            "event_count": 5123456,
            "sinks": {
                "gcp_pubsub": {"status": "healthy", "lag_ms": 28},
                "aws_kinesis": {"status": "healthy", "lag_ms": 67}
            }
        }
    },
    "cross_cloud_consistency": {
        "sync_status": "synchronized",
        "max_lag_ms": 67,
        "divergence_events": 0
    }
}
*/

Conclusion

HeliosDB's Change Data Capture with Event Streaming provides native, high-performance data synchronization without external tools or source database impact.

Key Value Propositions:

• Zero source impact - Asynchronous WAL-based capture
• Exactly-once semantics - Guaranteed delivery
• Schema evolution - Automatic compatibility handling
• Multi-cloud support - Kafka, Kinesis, Pulsar, EventHub

Call to Action: Enable CDC with SELECT helios_cdc_enable(table_name => 'your_table');