Reconcile MySQL to PostgreSQL Migration Parity and Synchronize Sequences for Cutover

id: mysql-to-postgres-migration-parity-and-sequence-cutover
namespace: company.database

description: |
  Audit heterogeneous replication parity between MySQL and PostgreSQL, detect table
  row count drift using DuckDB, and advance PostgreSQL serial sequences to match MySQL
  auto-increment values before final application cutover.

inputs:
  - id: mysql_host
    type: STRING
    description: Hostname or IP address of the source MySQL database.
    defaults: mysql.internal.corp

  - id: mysql_port
    type: INT
    description: Network port of the source MySQL database.
    defaults: 3306

  - id: mysql_database
    type: STRING
    description: Schema or database name in MySQL holding replicated tables.
    defaults: production_db

  - id: mysql_user
    type: STRING
    description: Database user for MySQL metadata inspection queries.
    defaults: migration_reader

  - id: postgres_host
    type: STRING
    description: Hostname or IP address of the target PostgreSQL database.
    defaults: postgres.internal.corp

  - id: postgres_port
    type: INT
    description: Network port of the target PostgreSQL database.
    defaults: 5432

  - id: postgres_database
    type: STRING
    description: Database name on the target PostgreSQL server.
    defaults: production_db

  - id: postgres_schema
    type: STRING
    description: Target PostgreSQL schema containing the replicated tables.
    defaults: public

  - id: postgres_user
    type: STRING
    description: Database user with sequence administration privileges on PostgreSQL.
    defaults: migration_admin

  - id: apply_sequence_cutover
    type: BOOL
    description: When true and dry_run is false, advances PostgreSQL sequence values
      using setval.
    defaults: false

  - id: cutover_approval_timeout
    type: DURATION
    description: Maximum duration to pause for DBA review before proceeding with
      sequence cutover.
    defaults: PT15M

  - id: dry_run
    type: BOOL
    description: When true, inspects schemas and outputs reconciliation reports
      without modifying PostgreSQL sequences.
    defaults: true

triggers:
  - id: scheduled_parity_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run periodic replication parity audit during the pre-cutover
      testing window.
    cron: 0 2 * * 0
    disabled: true

tasks:
  - id: list_mysql_tables
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Discover all base tables and auto-increment columns from the source
      MySQL schema.
    url: "jdbc:mysql://{{ inputs.mysql_host }}:{{ inputs.mysql_port }}/{{
      inputs.mysql_database }}?useSSL=false&allowPublicKeyRetrieval=true"
    username: "{{ inputs.mysql_user }}"
    password: "{{ secret('MYSQL_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT 
        t.TABLE_NAME AS table_name,
        c.COLUMN_NAME AS auto_increment_column
      FROM information_schema.TABLES t
      LEFT JOIN information_schema.COLUMNS c 
        ON t.TABLE_SCHEMA = c.TABLE_SCHEMA 
        AND t.TABLE_NAME = c.TABLE_NAME 
        AND c.EXTRA LIKE '%auto_increment%'
      WHERE t.TABLE_SCHEMA = '{{ inputs.mysql_database }}'
        AND t.TABLE_TYPE = 'BASE TABLE'
      ORDER BY t.TABLE_NAME;

  - id: exact_count_mysql_tables
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Compute exact row counts and MAX auto-increment high-water marks
      across discovered MySQL tables.
    url: "jdbc:mysql://{{ inputs.mysql_host }}:{{ inputs.mysql_port }}/{{
      inputs.mysql_database }}?useSSL=false&allowPublicKeyRetrieval=true"
    username: "{{ inputs.mysql_user }}"
    password: "{{ secret('MYSQL_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      {% for r in outputs.list_mysql_tables.rows %}
      SELECT 
        '{{ r.table_name }}' AS table_name, 
        count(*) AS exact_row_count, 
        {% if r.auto_increment_column %}
        COALESCE(max(`{{ r.auto_increment_column }}`), 0) + 1 AS auto_increment,
        '{{ r.auto_increment_column }}' AS auto_increment_column
        {% else %}
        NULL AS auto_increment,
        NULL AS auto_increment_column
        {% endif %}
      FROM `{{ inputs.mysql_database }}`.`{{ r.table_name }}`
      {% if not loop.last %} UNION ALL {% endif %}
      {% endfor %}

  - id: inspect_postgres_catalog
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Discover all target PostgreSQL tables and resolve sequence linkages
      for serial and identity columns.
    url: "jdbc:postgresql://{{ inputs.postgres_host }}:{{ inputs.postgres_port }}/{{
      inputs.postgres_database }}"
    username: "{{ inputs.postgres_user }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT DISTINCT ON (t.table_name)
        t.table_name,
        c.column_name AS pk_column,
        pg_get_serial_sequence(quote_ident(t.table_schema) || '.' || quote_ident(t.table_name), c.column_name) AS sequence_name,
        COALESCE(seq.last_value, 1) AS last_value
      FROM information_schema.tables t
      LEFT JOIN information_schema.columns c 
        ON t.table_schema = c.table_schema 
        AND t.table_name = c.table_name
        AND (
          c.column_default LIKE 'nextval(%' 
          OR c.is_identity = 'YES' 
          OR pg_get_serial_sequence(quote_ident(t.table_schema) || '.' || quote_ident(t.table_name), c.column_name) IS NOT NULL
        )
      LEFT JOIN pg_sequences seq 
        ON seq.schemaname = t.table_schema 
        AND seq.sequencename = replace(pg_get_serial_sequence(quote_ident(t.table_schema) || '.' || quote_ident(t.table_name), c.column_name), quote_ident(t.table_schema) || '.', '')
      WHERE t.table_schema = '{{ inputs.postgres_schema }}'
        AND t.table_type = 'BASE TABLE'
      ORDER BY t.table_name, c.column_name;

  - id: exact_count_postgres_tables
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Compute exact live row counts across all target PostgreSQL tables
      using a unified UNION ALL query.
    url: "jdbc:postgresql://{{ inputs.postgres_host }}:{{ inputs.postgres_port }}/{{
      inputs.postgres_database }}"
    username: "{{ inputs.postgres_user }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      {% for r in outputs.inspect_postgres_catalog.rows %}
      SELECT '{{ r.table_name }}' AS table_name, count(*) AS exact_row_count FROM "{{ inputs.postgres_schema }}"."{{ r.table_name }}"
      {% if not loop.last %} UNION ALL {% endif %}
      {% endfor %}

  - id: reconcile_migration_parity
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Join MySQL and PostgreSQL metadata in DuckDB to flag row drift,
      missing tables, and lagging sequences.
    inputFiles:
      mysql_tables.json: "{{ outputs.exact_count_mysql_tables.rows | toJson }}"
      postgres_meta.json: "{{ outputs.inspect_postgres_catalog.rows | toJson }}"
      postgres_counts.json: "{{ outputs.exact_count_postgres_tables.rows | toJson }}"
    outputFiles:
      - migration_parity_reconciliation.csv
    sql: |
      CREATE TABLE mysql_src AS SELECT * FROM read_json_auto('{{ workingDir }}/mysql_tables.json');
      CREATE TABLE pg_meta AS SELECT * FROM read_json_auto('{{ workingDir }}/postgres_meta.json');
      CREATE TABLE pg_counts AS SELECT * FROM read_json_auto('{{ workingDir }}/postgres_counts.json');

      CREATE TABLE pg_target AS
      SELECT 
        m.table_name,
        m.pk_column,
        m.sequence_name,
        m.last_value AS sequence_last_val,
        c.exact_row_count
      FROM pg_meta m
      JOIN pg_counts c ON m.table_name = c.table_name;

      CREATE TABLE reconciliation AS
      SELECT 
        COALESCE(m.table_name, p.table_name) AS table_name,
        p.pk_column,
        COALESCE(m.exact_row_count, 0) AS mysql_rows,
        COALESCE(p.exact_row_count, 0) AS postgres_rows,
        TRY_CAST(m.auto_increment AS BIGINT) AS mysql_auto_increment,
        p.sequence_name AS postgres_sequence,
        CASE WHEN p.sequence_name IS NOT NULL THEN p.sequence_last_val ELSE NULL END AS postgres_sequence_val,
        CASE 
          WHEN m.table_name IS NULL THEN 'MISSING_IN_MYSQL'
          WHEN p.table_name IS NULL THEN 'MISSING_IN_POSTGRES'
          WHEN m.exact_row_count != p.exact_row_count THEN 'ROW_COUNT_MISMATCH'
          WHEN m.auto_increment IS NOT NULL AND p.sequence_name IS NOT NULL AND (p.sequence_last_val IS NULL OR p.sequence_last_val < TRY_CAST(m.auto_increment AS BIGINT) - 1) THEN 'SEQUENCE_OUT_OF_SYNC'
          ELSE 'IN_SYNC'
        END AS parity_status,
        CASE
          WHEN m.auto_increment IS NOT NULL AND p.sequence_name IS NOT NULL AND (p.sequence_last_val IS NULL OR p.sequence_last_val < TRY_CAST(m.auto_increment AS BIGINT) - 1)
          THEN CASE 
            WHEN TRY_CAST(m.auto_increment AS BIGINT) <= 1 THEN 'SELECT setval(''' || p.sequence_name || ''', 1, false);'
            ELSE 'SELECT setval(''' || p.sequence_name || ''', ' || CAST(TRY_CAST(m.auto_increment AS BIGINT) - 1 AS VARCHAR) || ', true);'
          END
          ELSE NULL
        END AS fix_sequence_sql
      FROM mysql_src m
      FULL OUTER JOIN pg_target p ON m.table_name = p.table_name;

      COPY reconciliation TO '{{ workingDir }}/migration_parity_reconciliation.csv' (HEADER, DELIMITER ',');

      SELECT 
        count(*) AS total_tables_audited,
        count(*) FILTER (WHERE parity_status = 'ROW_COUNT_MISMATCH') AS row_mismatches,
        count(*) FILTER (WHERE parity_status = 'SEQUENCE_OUT_OF_SYNC') AS unsynced_sequences,
        count(*) FILTER (WHERE parity_status = 'MISSING_IN_POSTGRES') AS missing_target_tables,
        count(*) FILTER (WHERE parity_status = 'IN_SYNC') AS verified_synced_tables
      FROM reconciliation;

      SELECT table_name, fix_sequence_sql FROM reconciliation WHERE fix_sequence_sql IS NOT NULL;
    fetchType: FETCH

  - id: sync_postgres_sequences
    type: io.kestra.plugin.core.flow.If
    description: Pause for DBA review and apply generated sequence alignment
      statements to PostgreSQL when cutover is enabled.
    condition: "{{
      outputs.reconcile_migration_parity.outputs[0].rows[0].unsynced_sequences >
      0 and inputs.apply_sequence_cutover == true and inputs.dry_run == false
      }}"
    then:
      - id: pause_cutover_approval
        type: io.kestra.plugin.core.flow.Pause
        description: Pause execution to allow DBAs to review reconciliation metrics
          before executing sequence updates.
        pauseDuration: "{{ inputs.cutover_approval_timeout }}"
        behavior: FAIL

      - id: execute_sequence_cutover
        type: io.kestra.plugin.jdbc.postgresql.Queries
        description: Advance PostgreSQL sequence counters to match source MySQL
          auto-increment values.
        url: "jdbc:postgresql://{{ inputs.postgres_host }}:{{ inputs.postgres_port }}/{{
          inputs.postgres_database }}"
        username: "{{ inputs.postgres_user }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        sql: |
          {% for r in outputs.reconcile_migration_parity.outputs[1].rows %}
          {{ r.fix_sequence_sql }}
          {% endfor %}

      - id: log_sequence_cutover
        type: io.kestra.plugin.core.log.Log
        description: Log successful sequence update counts to execution audit logs.
        message: "Successfully synchronized {{
          outputs.reconcile_migration_parity.outputs[0].rows[0].unsynced_sequen\
          ces }} sequences on target PostgreSQL database."
    else:
      - id: log_cutover_status
        type: io.kestra.plugin.core.log.Log
        description: Log informational message when sequence synchronization is skipped
          or unnecessary.
        message: "Sequence cutover check complete: {{
          outputs.reconcile_migration_parity.outputs[0].rows[0].unsynced_sequen\
          ces }} sequences require alignment (dry_run: {{ inputs.dry_run }},
          apply_cutover: {{ inputs.apply_sequence_cutover }})."

  - id: certify_cutover_gate
    type: io.kestra.plugin.core.execution.Assert
    description: Assert zero row drift and verified sequence alignment before
      authorizing application cutover.
    conditions:
      - "{{ inputs.dry_run == true or
        (outputs.reconcile_migration_parity.outputs[0].rows[0].row_mismatches ==
        0 and
        outputs.reconcile_migration_parity.outputs[0].rows[0].missing_target_ta\
        bles == 0) }}"
    errorMessage: "Migration cutover gate rejected: Row count mismatch or missing
      target tables detected between MySQL and PostgreSQL."

  - id: dispatch_cutover_report
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatch migration reconciliation status and cutover readiness
      report to the DBA team channel.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "#database-migrations",
        "text": ":satellite: *MySQL to PostgreSQL Migration Parity Report*\nTotal Tables: `{{ outputs.reconcile_migration_parity.outputs[0].rows[0].total_tables_audited }}`\nVerified Synced: `{{ outputs.reconcile_migration_parity.outputs[0].rows[0].verified_synced_tables }}`\nRow Mismatches: `{{ outputs.reconcile_migration_parity.outputs[0].rows[0].row_mismatches }}`\nUnsynced Sequences: `{{ outputs.reconcile_migration_parity.outputs[0].rows[0].unsynced_sequences }}`\nReconciliation CSV: `{{ outputs.reconcile_migration_parity.outputFiles['migration_parity_reconciliation.csv'] }}`"
      }