+------------------------------------------------------------------+
|  Flink CDC End-to-End Data Flow                                   |
|                                                                   |
|  MySQL/PG/Oracle                                                  |
|       |                                                           |
|       | binlog / WAL / redo log                                   |
|       v                                                           |
|  Debezium Connector (Flink Source)                                |
|       |                                                           |
|       | ChangeEvent (JSON / Avro)                                 |
|       v                                                           |
|  Flink Stream Processing                                          |
|       |-- Schema Mapping (source columns -> Ontology properties)  |
|       |-- Data Cleansing (type conversion, null handling)          |
|       |-- Deduplication (primary-key based)                        |
|       |-- Enrichment (join dimension tables)                       |
|       v                                                           |
|  Dual Write                                                       |
|       |-- Iceberg Table (via Nessie catalog)                      |
|       |-- Doris Instance Table (real-time query layer)            |
|       v                                                           |
|  Ontology Event Bus                                               |
|       |-- InstanceCreatedEvent                                     |
|       |-- InstanceUpdatedEvent                                     |
|       |-- InstanceDeletedEvent                                     |
+------------------------------------------------------------------+

+------------------------------------------------------------------+
|                    Flink CDC Pipeline Architecture                 |
|                                                                   |
|  +-----------------+  +-------------------+  +-----------------+  |
|  | Source Layer     |  | Processing Layer  |  | Sink Layer      |  |
|  |                 |  |                   |  |                 |  |
|  | MySQLSource     |  | SchemaMapper      |  | IcebergSink     |  |
|  | PostgresSource  |  | DataCleanser      |  | DorisSink       |  |
|  | OracleSource    |  | Deduplicator      |  | EventBusSink    |  |
|  | MongoDBSource   |  | Enricher          |  | MetricsSink     |  |
|  |                 |  | Router            |  |                 |  |
|  +-----------------+  +-------------------+  +-----------------+  |
|                                                                   |
|  Cross-Cutting Concerns:                                          |
|  +------------------------------------------------------------+  |
|  | Checkpoint Mgr | Schema Registry | Error Handler | Metrics  |  |
|  +------------------------------------------------------------+  |
+------------------------------------------------------------------+

from pydantic import BaseModel, Field
from typing import Optional, List, Dict
from enum import Enum


class SourceType(str, Enum):
    MYSQL = "mysql"
    POSTGRESQL = "postgresql"
    ORACLE = "oracle"
    MONGODB = "mongodb"
    SQLSERVER = "sqlserver"


class CDCPipelineConfig(BaseModel):
    """CDC Pipeline configuration"""
    pipeline_id: str
    display_name: str
    source: SourceConfig
    processing: ProcessingConfig
    sink: SinkConfig
    checkpoint: CheckpointConfig = CheckpointConfig()
    error_handling: ErrorHandlingConfig = ErrorHandlingConfig()


class SourceConfig(BaseModel):
    """Data source configuration"""
    source_type: SourceType
    hostname: str
    port: int
    database: str
    tables: List[str]                    # Tables to monitor
    username: str
    password_secret_ref: str             # K8s Secret reference
    server_id: Optional[int] = None      # MySQL server-id
    slot_name: Optional[str] = None      # PostgreSQL replication slot
    startup_mode: str = "initial"        # initial | latest-offset | timestamp
    startup_timestamp: Optional[int] = None


class ProcessingConfig(BaseModel):
    """Processing layer configuration"""
    schema_mapping: Dict[str, SchemaMapping]  # table name -> mapping rules
    deduplication: DeduplicationConfig = DeduplicationConfig()
    enrichment: Optional[EnrichmentConfig] = None


class SchemaMapping(BaseModel):
    """Schema mapping: source table -> Ontology ObjectType"""
    source_table: str
    target_object_type_rid: str
    field_mappings: Dict[str, FieldMapping]  # source field -> target property
    primary_key_field: str
    rid_template: str = "ri.{object_type}.main.{pk}"


class FieldMapping(BaseModel):
    """Field mapping"""
    source_column: str
    target_property: str
    type_conversion: Optional[str] = None   # Type conversion function
    default_value: Optional[str] = None
    nullable: bool = True


class SinkConfig(BaseModel):
    """Sink layer configuration"""
    iceberg_enabled: bool = True
    iceberg_catalog: str = "nessie"
    iceberg_warehouse: str = "s3://coomia-dip-warehouse/"
    doris_enabled: bool = True
    doris_fe_endpoints: List[str] = []
    event_bus_enabled: bool = True


class CheckpointConfig(BaseModel):
    """Checkpoint configuration"""
    interval_ms: int = 60000             # 1 minute
    min_pause_ms: int = 500
    timeout_ms: int = 600000             # 10 minutes
    max_concurrent: int = 1
    state_backend: str = "rocksdb"
    state_dir: str = "s3://coomia-dip-checkpoints/"


class ErrorHandlingConfig(BaseModel):
    """Error handling configuration"""
    max_retries: int = 3
    retry_interval_ms: int = 5000
    dead_letter_enabled: bool = True
    dead_letter_topic: str = "cdc-dead-letter"
    skip_corrupted: bool = False

class MySQLCDCSource:
    """MySQL CDC Source builder"""

    def build(self, config: SourceConfig) -> FlinkSource:
        """Build MySQL CDC Source"""
        return MySqlSource.builder() \
            .hostname(config.hostname) \
            .port(config.port) \
            .database_list(config.database) \
            .table_list(*config.tables) \
            .username(config.username) \
            .password(self._resolve_secret(config.password_secret_ref)) \
            .server_id(config.server_id or self._generate_server_id()) \
            .deserializer(JsonDebeziumDeserializationSchema()) \
            .startup_options(self._build_startup_options(config)) \
            .build()

    def _build_startup_options(self, config: SourceConfig):
        if config.startup_mode == "initial":
            return StartupOptions.initial()
        elif config.startup_mode == "latest-offset":
            return StartupOptions.latest_offset()
        elif config.startup_mode == "timestamp":
            return StartupOptions.timestamp(config.startup_timestamp)
        else:
            raise ValueError(f"Unknown startup mode: {config.startup_mode}")

{
  "before": {
    "id": 1001,
    "name": "Pump-A",
    "status": "running",
    "temperature": 72.5
  },
  "after": {
    "id": 1001,
    "name": "Pump-A",
    "status": "maintenance",
    "temperature": 85.3
  },
  "source": {
    "connector": "mysql",
    "db": "factory_db",
    "table": "equipment",
    "ts_ms": 1711267200000,
    "server_id": 1,
    "file": "mysql-bin.000003",
    "pos": 12345
  },
  "op": "u",
  "ts_ms": 1711267200123
}

+------------------------------------------------------------------+
|  Full Snapshot + Incremental Sync Flow                            |
|                                                                   |
|  Phase 1: Snapshot (Full Snapshot)                                |
|  ───────────────────────────                                      |
|  1. Acquire global read lock (FLUSH TABLES WITH READ LOCK)        |
|  2. Record current binlog position                                |
|  3. Read table schemas (SHOW CREATE TABLE)                        |
|  4. Release read lock                                             |
|  5. Read full data in chunks (SELECT * WHERE pk BETWEEN ? AND ?)  |
|  6. Each chunk produces op=r events                               |
|                                                                   |
|  Phase 2: Incremental (Incremental Sync)                          |
|  ───────────────────────────                                      |
|  1. Start consuming from the binlog position recorded in snapshot |
|  2. Receive INSERT/UPDATE/DELETE events in real time              |
|  3. Run continuously until the pipeline is stopped                |
+------------------------------------------------------------------+

class SchemaMapper:
    """Schema mapper: source table fields -> Ontology properties"""

    def __init__(self, mappings: Dict[str, SchemaMapping]):
        self._mappings = mappings

    def map_event(self, event: ChangeEvent) -> Optional[OntologyChangeEvent]:
        """Convert a Debezium ChangeEvent to an OntologyChangeEvent"""
        table_name = f"{event.source.db}.{event.source.table}"
        mapping = self._mappings.get(table_name)
        if mapping is None:
            return None  # Skip tables without configured mappings

        if event.op in ("c", "r"):
            return self._map_create(event, mapping)
        elif event.op == "u":
            return self._map_update(event, mapping)
        elif event.op == "d":
            return self._map_delete(event, mapping)
        return None

    def _map_create(self, event: ChangeEvent, mapping: SchemaMapping) -> OntologyChangeEvent:
        properties = {}
        for source_col, field_map in mapping.field_mappings.items():
            value = event.after.get(source_col)
            if value is None and field_map.default_value is not None:
                value = field_map.default_value
            if field_map.type_conversion:
                value = self._convert_type(value, field_map.type_conversion)
            properties[field_map.target_property] = value

        pk_value = event.after[mapping.primary_key_field]
        instance_rid = mapping.rid_template.format(
            object_type=mapping.target_object_type_rid.split(".")[-1],
            pk=pk_value
        )

        return OntologyChangeEvent(
            operation="create",
            object_type_rid=mapping.target_object_type_rid,
            instance_rid=instance_rid,
            properties=properties,
            source_timestamp=event.source.ts_ms,
            source_table=f"{event.source.db}.{event.source.table}"
        )

    def _map_update(self, event: ChangeEvent, mapping: SchemaMapping) -> OntologyChangeEvent:
        # Only map changed fields
        changed_properties = {}
        for source_col, field_map in mapping.field_mappings.items():
            old_value = event.before.get(source_col) if event.before else None
            new_value = event.after.get(source_col) if event.after else None
            if old_value != new_value:
                if field_map.type_conversion:
                    new_value = self._convert_type(new_value, field_map.type_conversion)
                changed_properties[field_map.target_property] = new_value

        pk_value = event.after[mapping.primary_key_field]
        instance_rid = mapping.rid_template.format(
            object_type=mapping.target_object_type_rid.split(".")[-1],
            pk=pk_value
        )

        return OntologyChangeEvent(
            operation="update",
            object_type_rid=mapping.target_object_type_rid,
            instance_rid=instance_rid,
            properties=changed_properties,
            source_timestamp=event.source.ts_ms,
            source_table=f"{event.source.db}.{event.source.table}"
        )

    def _convert_type(self, value, conversion: str):
        """Type conversion"""
        converters = {
            "to_string": str,
            "to_int": int,
            "to_float": float,
            "to_boolean": lambda v: v in (1, True, "true", "yes"),
            "epoch_ms_to_datetime": lambda v: datetime.fromtimestamp(v / 1000),
            "cents_to_dollars": lambda v: v / 100.0 if v else None,
        }
        converter = converters.get(conversion)
        if converter and value is not None:
            return converter(value)
        return value

class Deduplicator:
    """Primary-key and event-time based deduplicator"""

    def __init__(self, state_backend, dedup_window_ms: int = 60000):
        self._state = state_backend
        self._window_ms = dedup_window_ms

    async def is_duplicate(self, event: OntologyChangeEvent) -> bool:
        """Check if the event is a duplicate"""
        key = f"{event.object_type_rid}:{event.instance_rid}"
        last_ts = await self._state.get(key)

        if last_ts is not None and event.source_timestamp <= last_ts:
            return True  # Duplicate event

        await self._state.put(key, event.source_timestamp)
        return False

    async def cleanup_expired(self):
        """Clean up expired deduplication state"""
        cutoff = int(time.time() * 1000) - self._window_ms
        await self._state.remove_before(cutoff)

class Enricher:
    """Data enricher: join dimension tables to add fields"""

    def __init__(self, dimension_sources: Dict[str, DimensionSource]):
        self._sources = dimension_sources

    async def enrich(
        self,
        event: OntologyChangeEvent,
        enrichment_config: EnrichmentConfig
    ) -> OntologyChangeEvent:
        for rule in enrichment_config.rules:
            source = self._sources.get(rule.dimension_source)
            if source is None:
                continue

            lookup_key = event.properties.get(rule.lookup_field)
            if lookup_key is None:
                continue

            dimension_data = await source.lookup(lookup_key)
            if dimension_data:
                for target_field, source_field in rule.field_mappings.items():
                    event.properties[target_field] = dimension_data.get(source_field)

        return event

class IcebergSinkWriter:
    """Iceberg Sink: write to Iceberg tables"""

    def __init__(self, catalog_config: dict):
        self._catalog = NessieCatalog(**catalog_config)

    async def write(self, event: OntologyChangeEvent):
        table_name = self._resolve_table_name(event.object_type_rid)
        table = self._catalog.load_table(table_name)

        if event.operation == "create":
            record = self._to_iceberg_record(event)
            table.append(record)
        elif event.operation == "update":
            # Iceberg uses Merge-on-Read strategy
            table.merge(
                key=event.instance_rid,
                updates=event.properties
            )
        elif event.operation == "delete":
            table.delete(
                filter=f"rid = '{event.instance_rid}'"
            )

    async def commit(self, checkpoint_id: int):
        """Commit Iceberg transaction at checkpoint time"""
        self._catalog.commit(
            message=f"CDC checkpoint {checkpoint_id}",
            branch="main"
        )

class DorisSinkWriter:
    """Doris Sink: Stream Load writes"""

    def __init__(self, fe_endpoints: List[str]):
        self._endpoints = fe_endpoints
        self._buffer: List[dict] = []
        self._buffer_size = 1000
        self._flush_interval_ms = 5000

    async def write(self, event: OntologyChangeEvent):
        row = self._to_doris_row(event)
        self._buffer.append(row)

        if len(self._buffer) >= self._buffer_size:
            await self._flush()

    async def _flush(self):
        if not self._buffer:
            return

        table_name = self._resolve_table_name(self._buffer[0])
        csv_data = self._to_csv(self._buffer)

        endpoint = self._select_endpoint()
        response = await self._stream_load(
            endpoint=endpoint,
            table=table_name,
            data=csv_data,
            format="csv",
            merge_type="MERGE",  # UPSERT semantics
            delete_condition="__deleted = 1"
        )

        if response.status != "Success":
            raise DorisSinkError(f"Stream Load failed: {response.message}")

        self._buffer.clear()

    async def _stream_load(self, endpoint, table, data, **kwargs):
        """Call Doris Stream Load API"""
        url = f"http://{endpoint}/api/{table}/_stream_load"
        headers = {
            "Content-Type": "text/csv",
            "format": kwargs.get("format", "csv"),
            "merge_type": kwargs.get("merge_type", "APPEND"),
        }
        async with aiohttp.ClientSession() as session:
            async with session.put(url, data=data, headers=headers) as resp:
                return await resp.json()

+------------------------------------------------------------------+
|  Dual-Write Consistency Strategy                                  |
|                                                                   |
|  Flink Checkpoint                                                 |
|       |                                                           |
|       v                                                           |
|  Phase 1: Pre-commit                                              |
|       |-- Iceberg: prepare commit (write data files)              |
|       |-- Doris: buffer ready                                     |
|       v                                                           |
|  Phase 2: Commit                                                  |
|       |-- Iceberg: commit transaction (atomic operation)          |
|       |-- Doris: Stream Load commit                               |
|       v                                                           |
|  Phase 3: Post-commit                                             |
|       |-- Publish events to Event Bus                             |
|       |-- Update pipeline status                                  |
|                                                                   |
|  Failure Recovery:                                                |
|  If Doris write fails but Iceberg succeeds:                       |
|  -> Trigger Doris compensating task to reload from Iceberg        |
+------------------------------------------------------------------+

class SchemaEvolutionHandler:
    """Schema Evolution handler"""

    def __init__(self, schema_registry, iceberg_catalog):
        self._schema_registry = schema_registry
        self._iceberg = iceberg_catalog

    async def handle_ddl(self, ddl_event: DDLEvent):
        """Handle DDL change events"""
        if ddl_event.type == "ALTER_TABLE_ADD_COLUMN":
            await self._handle_add_column(ddl_event)
        elif ddl_event.type == "ALTER_TABLE_MODIFY_COLUMN":
            await self._handle_modify_column(ddl_event)
        elif ddl_event.type == "ALTER_TABLE_DROP_COLUMN":
            await self._handle_drop_column(ddl_event)

    async def _handle_add_column(self, ddl_event: DDLEvent):
        """Handle column addition"""
        mapping = self._get_mapping(ddl_event.table)
        if mapping is None:
            return

        # 1. Update Ontology Schema (register new property)
        new_property = PropertyDefinition(
            api_name=ddl_event.column_name,
            display_name=ddl_event.column_name,
            property_type=self._map_sql_type(ddl_event.column_type),
            nullable=ddl_event.nullable
        )
        await self._schema_registry.add_property(
            mapping.target_object_type_rid,
            new_property
        )

        # 2. Update Iceberg table schema
        table = self._iceberg.load_table(
            self._resolve_table_name(mapping.target_object_type_rid)
        )
        table.update_schema() \
            .add_column(ddl_event.column_name, self._to_iceberg_type(ddl_event.column_type)) \
            .commit()

        # 3. Update field mapping
        mapping.field_mappings[ddl_event.column_name] = FieldMapping(
            source_column=ddl_event.column_name,
            target_property=ddl_event.column_name
        )

    async def _handle_modify_column(self, ddl_event: DDLEvent):
        """Handle column type changes"""
        mapping = self._get_mapping(ddl_event.table)
        if mapping is None:
            return

        # Iceberg supports safe type promotions (int -> long, float -> double)
        table = self._iceberg.load_table(
            self._resolve_table_name(mapping.target_object_type_rid)
        )

        old_type = table.schema().find_field(ddl_event.column_name).field_type
        new_type = self._to_iceberg_type(ddl_event.column_type)

        if self._is_safe_promotion(old_type, new_type):
            table.update_schema() \
                .update_column(ddl_event.column_name, new_type) \
                .commit()
        else:
            await self._alert_unsafe_schema_change(ddl_event)

+------------------------------------------------------------------+
|  Exactly-Once Semantics Implementation                            |
|                                                                   |
|  Component             Mechanism                                  |
|  ──────────           ──────────────────────                      |
|  Source (Debezium)     binlog offset persisted in Flink State      |
|  Processing (Flink)    Checkpoint Barrier alignment                |
|  Sink (Iceberg)        Two-Phase Commit (2PC)                      |
|  Sink (Doris)          Idempotent UPSERT (primary key dedup)       |
|                                                                   |
|  Checkpoint Flow:                                                  |
|  1. JobManager injects Checkpoint Barrier                          |
|  2. Source records current binlog offset                            |
|  3. Barrier flows through all operators, each persists local state |
|  4. Sink executes Pre-commit (write but don't commit)              |
|  5. After all operators complete, JobManager signals Commit        |
|  6. Sink executes Commit (atomic commit)                           |
+------------------------------------------------------------------+

class CDCPipelineRecovery:
    """CDC Pipeline failure recovery"""

    async def recover(self, pipeline_id: str):
        """Recover pipeline from the most recent checkpoint"""
        # 1. Get the latest checkpoint
        checkpoint = await self._checkpoint_store.get_latest(pipeline_id)
        if checkpoint is None:
            raise RecoveryError(f"No checkpoint found for pipeline {pipeline_id}")

        # 2. Restore Source binlog position
        binlog_offset = checkpoint.source_state["binlog_offset"]

        # 3. Restore Processing state (deduplication state, etc.)
        processing_state = checkpoint.processing_state

        # 4. Restore Sink transaction state
        # Iceberg: rollback uncommitted transactions
        await self._iceberg_sink.rollback_pending()
        # Doris: idempotent replay, no special handling needed

        # 5. Restart pipeline from checkpoint position
        await self._restart_pipeline(
            pipeline_id,
            binlog_offset=binlog_offset,
            processing_state=processing_state
        )

    async def _restart_pipeline(self, pipeline_id, binlog_offset, processing_state):
        config = await self._config_store.get(pipeline_id)
        config.source.startup_mode = "specific-offset"
        config.source.startup_offset = binlog_offset

        pipeline = CDCPipelineBuilder(config).build()
        pipeline.restore_state(processing_state)
        await pipeline.start()

class CDCPipelineMetrics:
    """CDC Pipeline monitoring metrics"""

    def __init__(self, metrics_registry):
        self._registry = metrics_registry

        # Source metrics
        self.source_events_total = self._registry.counter(
            "cdc_source_events_total",
            labels=["pipeline_id", "table", "operation"]
        )
        self.source_lag_ms = self._registry.gauge(
            "cdc_source_lag_ms",
            labels=["pipeline_id"]
        )

        # Processing metrics
        self.processing_latency_ms = self._registry.histogram(
            "cdc_processing_latency_ms",
            labels=["pipeline_id", "stage"]
        )
        self.dedup_hit_total = self._registry.counter(
            "cdc_dedup_hit_total",
            labels=["pipeline_id"]
        )

        # Sink metrics
        self.sink_write_total = self._registry.counter(
            "cdc_sink_write_total",
            labels=["pipeline_id", "sink_type", "status"]
        )
        self.sink_latency_ms = self._registry.histogram(
            "cdc_sink_latency_ms",
            labels=["pipeline_id", "sink_type"]
        )

        # Checkpoint metrics
        self.checkpoint_duration_ms = self._registry.histogram(
            "cdc_checkpoint_duration_ms",
            labels=["pipeline_id"]
        )
        self.checkpoint_size_bytes = self._registry.gauge(
            "cdc_checkpoint_size_bytes",
            labels=["pipeline_id"]
        )

class CDCLagMonitor:
    """CDC lag monitor"""

    def __init__(self, alert_threshold_ms: int = 30000):
        self._threshold_ms = alert_threshold_ms

    async def check_lag(self, pipeline_id: str, event: OntologyChangeEvent):
        current_ms = int(time.time() * 1000)
        lag_ms = current_ms - event.source_timestamp

        metrics.source_lag_ms.set(lag_ms, pipeline_id=pipeline_id)

        if lag_ms > self._threshold_ms:
            await self._send_alert(
                pipeline_id=pipeline_id,
                lag_ms=lag_ms,
                threshold_ms=self._threshold_ms,
                message=f"CDC pipeline {pipeline_id} lag {lag_ms}ms exceeds threshold"
            )

class CDCPipelineManager:
    """CDC Pipeline lifecycle management"""

    async def create(self, config: CDCPipelineConfig) -> str:
        """Create a pipeline"""
        await self._validate_config(config)
        await self._test_connection(config.source)
        pipeline_id = await self._store.save(config)
        return pipeline_id

    async def start(self, pipeline_id: str):
        """Start a pipeline"""
        config = await self._store.get(pipeline_id)
        pipeline = CDCPipelineBuilder(config).build()
        await pipeline.start()
        await self._store.update_status(pipeline_id, "RUNNING")

    async def stop(self, pipeline_id: str):
        """Stop a pipeline (saves checkpoint)"""
        pipeline = self._running.get(pipeline_id)
        if pipeline:
            await pipeline.stop_with_savepoint()
            await self._store.update_status(pipeline_id, "STOPPED")

    async def restart(self, pipeline_id: str):
        """Restart a pipeline (recovers from last checkpoint)"""
        await self.stop(pipeline_id)
        await self.start(pipeline_id)

    async def get_status(self, pipeline_id: str) -> PipelineStatus:
        """Get pipeline status"""
        config = await self._store.get(pipeline_id)
        metrics = await self._metrics.get_pipeline_metrics(pipeline_id)
        return PipelineStatus(
            pipeline_id=pipeline_id,
            status=config.status,
            source_lag_ms=metrics.source_lag_ms,
            events_per_second=metrics.events_per_second,
            last_checkpoint_at=metrics.last_checkpoint_at,
            error_count=metrics.error_count
        )

apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-cdc-jobmanager
  namespace: coomia-dip
spec:
  replicas: 1
  template:
    spec:
      containers:
      - name: flink-jobmanager
        image: coomia-dip/flink-cdc:1.18
        args: ["jobmanager"]
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"
        env:
        - name: FLINK_PROPERTIES
          value: |
            jobmanager.rpc.address: flink-cdc-jobmanager
            state.backend: rocksdb
            state.checkpoints.dir: s3://coomia-dip-checkpoints/
            execution.checkpointing.interval: 60000
            execution.checkpointing.min-pause: 500
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-cdc-taskmanager
  namespace: coomia-dip
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: flink-taskmanager
        image: coomia-dip/flink-cdc:1.18
        args: ["taskmanager"]
        resources:
          requests:
            memory: "4Gi"
            cpu: "2"
          limits:
            memory: "8Gi"
            cpu: "4"

+------------------------------------------------------------------+
|  Resource Planning Reference                                      |
|                                                                   |
|  Source DB Scale       TPS     TaskManagers    Memory/TM  Parallel |
|  ──────────────      ─────   ──────────────  ────────   ──────── |
|  Small (<50 tables)   <1K     2               4GB        4        |
|  Medium (50-200)      1K-5K   4               8GB        8        |
|  Large (>200 tables)  >5K     8+              16GB       16       |
+------------------------------------------------------------------+

+------------------------------------------------------------------+
|  Data Integration Capability Comparison                           |
|                                                                   |
|  Capability            Palantir Magritte     coomia-dip Flink CDC  |
|  ──────────────       ──────────────────    ──────────────        |
|  CDC support           Yes (proprietary)     Yes (open source)    |
|  MySQL/PG/Oracle       Yes                   Yes                  |
|  MongoDB               Yes                   Yes                  |
|  Batch import          Yes                   Yes                  |
|  Schema Evolution      Yes                   Yes                  |
|  Exactly-Once          Yes                   Yes                  |
|  Latency               Sub-second            Sub-second           |
|  Custom transforms     Yes (code)            Yes (config+code)    |
|  Visual orchestration  Yes                   Planned              |
|  Open-source control   No                    Yes                  |
+------------------------------------------------------------------+

class TestCDCPipeline:

    @pytest.fixture
    async def mysql_container(self):
        """Start MySQL Testcontainer"""
        async with MySQLContainer("mysql:8.0") as mysql:
            yield mysql

    @pytest.mark.asyncio
    async def test_full_cdc_pipeline(self, mysql_container):
        """End-to-end test: MySQL changes -> Ontology instances"""
        # 1. Create source table and insert data
        await mysql_container.execute("""
            CREATE TABLE equipment (
                id INT PRIMARY KEY,
                name VARCHAR(100),
                status VARCHAR(50)
            )
        """)
        await mysql_container.execute(
            "INSERT INTO equipment VALUES (1, 'Pump-A', 'running')"
        )

        # 2. Start CDC Pipeline
        config = CDCPipelineConfig(
            pipeline_id="test-pipeline",
            source=SourceConfig(
                source_type=SourceType.MYSQL,
                hostname=mysql_container.host,
                port=mysql_container.port,
                database="test",
                tables=["test.equipment"],
                username="root",
                password_secret_ref="test-secret"
            ),
            processing=ProcessingConfig(
                schema_mapping={
                    "test.equipment": SchemaMapping(
                        source_table="test.equipment",
                        target_object_type_rid="ri.type.Equipment",
                        field_mappings={
                            "id": FieldMapping(source_column="id", target_property="equipment_id"),
                            "name": FieldMapping(source_column="name", target_property="display_name"),
                            "status": FieldMapping(source_column="status", target_property="status")
                        },
                        primary_key_field="id"
                    )
                }
            ),
            sink=SinkConfig(
                iceberg_enabled=False, doris_enabled=False, event_bus_enabled=True
            )
        )

        pipeline = CDCPipelineBuilder(config).build()
        events = []
        pipeline.add_listener(lambda e: events.append(e))
        await pipeline.start()

        # 3. Wait for snapshot completion
        await asyncio.sleep(5)

        # 4. Verify initial snapshot events
        assert len(events) == 1
        assert events[0].operation == "create"
        assert events[0].properties["display_name"] == "Pump-A"

        # 5. Execute UPDATE
        await mysql_container.execute(
            "UPDATE equipment SET status = 'maintenance' WHERE id = 1"
        )
        await asyncio.sleep(2)

        # 6. Verify incremental change events
        assert len(events) == 2
        assert events[1].operation == "update"
        assert events[1].properties["status"] == "maintenance"

        await pipeline.stop()