Dataset Schema Evolution & Compatibility Gate

id: dataset-schema-compatibility-gate
namespace: company.data

description: >
  Validate an incoming CSV dataset against an expected schema contract. The
  workflow detects added, removed, and type-changed columns, generates a
  structured compatibility audit with DuckDB, and blocks downstream processing
  when breaking schema changes are detected.

inputs:
  - id: expected_schema
    type: JSON
    defaults:
      columns:
        - name: customer_id
          type: integer
          required: true

        - name: name
          type: string
          required: true

        - name: email
          type: string
          required: true

        - name: city
          type: string
          required: false
    description: >
      Expected schema contract for the incoming dataset.

  - id: incoming_csv
    type: STRING
    defaults: |
      customer_id,name,email,city
      101,Amit,amit@example.com,Mumbai
      102,Riya,riya@example.com,Pune
    description: >
      CSV dataset whose detected schema will be validated against the expected
      schema contract.

triggers:
  - id: daily_schema_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 9 * * *"
    timezone: Asia/Kolkata

tasks:

  # ---------------------------------------------------------
  # 1. Detect the schema of the incoming CSV using DuckDB
  # ---------------------------------------------------------

  - id: inspect_dataset
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: >
      Reads the incoming CSV with DuckDB and detects the column names and data
      types using DuckDB schema inference.
    inputFiles:
      input.csv: "{{ inputs.incoming_csv }}"
    sql: |
      SELECT
        column_name,
        column_type
      FROM (
        DESCRIBE
        SELECT *
        FROM read_csv_auto(
          'input.csv',
          header = true
        )
      );
    fetchType: FETCH

  # ---------------------------------------------------------
  # 2. Compare detected schema against expected contract
  # ---------------------------------------------------------

  - id: compare_schemas
    type: io.kestra.plugin.scripts.python.Script
    dependencies:
      - kestra

    description: >
      Compares the schema detected from the incoming CSV against the expected
      schema contract and identifies compatible and breaking changes.

    script: |
      from kestra import Kestra

      expected = {{ inputs.expected_schema | toJson }}

      detected_rows = {{ outputs.inspect_dataset.rows | toJson }}

      # ---------------------------------------------------
      # Normalize DuckDB types into contract-level types
      # ---------------------------------------------------

      def normalize_type(data_type):
          data_type = str(data_type).upper().strip()

          integer_types = {
              "TINYINT",
              "SMALLINT",
              "INTEGER",
              "BIGINT",
              "HUGEINT",
              "UTINYINT",
              "USMALLINT",
              "UINTEGER",
              "UBIGINT"
          }

          float_types = {
              "FLOAT",
              "DOUBLE",
              "REAL",
              "DECIMAL",
              "NUMERIC"
          }

          string_types = {
              "VARCHAR",
              "CHAR",
              "TEXT",
              "STRING"
          }

          boolean_types = {
              "BOOLEAN",
              "BOOL"
          }

          date_types = {
              "DATE"
          }

          timestamp_types = {
              "TIMESTAMP",
              "TIMESTAMP WITH TIME ZONE",
              "TIMESTAMPTZ"
          }

          if data_type in integer_types:
              return "integer"

          if data_type in float_types:
              return "number"

          if data_type in string_types:
              return "string"

          if data_type in boolean_types:
              return "boolean"

          if data_type in date_types:
              return "date"

          if data_type in timestamp_types:
              return "timestamp"

          return data_type.lower()


      # ---------------------------------------------------
      # Convert expected schema into dictionary
      # ---------------------------------------------------

      expected_columns = {
          column["name"]: column
          for column in expected["columns"]
      }


      # ---------------------------------------------------
      # Convert DuckDB schema into dictionary
      # ---------------------------------------------------

      incoming_columns = {}

      for row in detected_rows:
          column_name = row["column_name"]
          column_type = row["column_type"]

          incoming_columns[column_name] = {
              "name": column_name,
              "type": normalize_type(column_type)
          }


      # ---------------------------------------------------
      # Detect added columns
      # ---------------------------------------------------

      added_columns = [
          name
          for name in incoming_columns
          if name not in expected_columns
      ]


      # ---------------------------------------------------
      # Detect removed columns
      # ---------------------------------------------------

      removed_columns = [
          name
          for name in expected_columns
          if name not in incoming_columns
      ]


      # ---------------------------------------------------
      # Detect type changes
      # ---------------------------------------------------

      type_changes = []

      for name in expected_columns:

          if name in incoming_columns:

              expected_type = normalize_type(
                  expected_columns[name]["type"]
              )

              incoming_type = normalize_type(
                  incoming_columns[name]["type"]
              )

              if expected_type != incoming_type:

                  type_changes.append({
                      "column": name,
                      "expected": expected_type,
                      "actual": incoming_type
                  })


      # ---------------------------------------------------
      # Build breaking changes
      # ---------------------------------------------------

      breaking_changes = []


      # Removed expected column = BREAKING
      for name in removed_columns:

          breaking_changes.append({
              "type": "REMOVED_COLUMN",
              "column": name
          })


      # Type change = BREAKING
      for change in type_changes:

          breaking_changes.append({
              "type": "TYPE_CHANGE",
              "column": change["column"],
              "expected": change["expected"],
              "actual": change["actual"]
          })


      # ---------------------------------------------------
      # Added columns are considered compatible.
      #
      # The CSV itself cannot reliably tell us whether a new
      # column is required or optional, so requiredness is
      # controlled by the expected schema contract.
      # ---------------------------------------------------

      compatible = len(breaking_changes) == 0


      # ---------------------------------------------------
      # Final comparison result
      # ---------------------------------------------------

      result = {
          "compatible": compatible,
          "added_columns": added_columns,
          "removed_columns": removed_columns,
          "type_changes": type_changes,
          "breaking_changes": breaking_changes
      }

      Kestra.outputs(result)

  # ---------------------------------------------------------
  # 3. Create an audit record using DuckDB
  # ---------------------------------------------------------

  - id: schema_audit
    type: io.kestra.plugin.jdbc.duckdb.Query

    description: >
      Creates a structured audit record containing the compatibility decision
      and counts of detected schema changes.

    inputFiles:
      schema_audit.csv: |
        compatible,added_columns_count,removed_columns_count,type_changes_count,breaking_changes_count
        "{{ outputs.compare_schemas.vars.compatible }}","{{ outputs.compare_schemas.vars.added_columns | length }}","{{ outputs.compare_schemas.vars.removed_columns | length }}","{{ outputs.compare_schemas.vars.type_changes | length }}","{{ outputs.compare_schemas.vars.breaking_changes | length }}"

    sql: |
      SELECT
        compatible,
        added_columns_count,
        removed_columns_count,
        type_changes_count,
        breaking_changes_count
      FROM read_csv_auto(
        'schema_audit.csv',
        header = true
      );

    fetchType: FETCH_ONE

  # ---------------------------------------------------------
  # 4. Compatibility gate
  # ---------------------------------------------------------

  - id: compatibility_gate
    type: io.kestra.plugin.core.flow.If

    description: >
      Allows downstream processing only when the incoming dataset is compatible
      with the expected schema contract.

    condition: "{{ outputs.compare_schemas.vars.compatible }}"

    then:

      # -----------------------------------------------------
      # SUCCESS PATH
      # -----------------------------------------------------

      - id: compatible_log
        type: io.kestra.plugin.core.log.Log

        description: >
          Logs the successful schema compatibility check.

        message: |
          ==========================================
          SCHEMA COMPATIBILITY CHECK PASSED
          ==========================================

          Added columns:
          {{ outputs.compare_schemas.vars.added_columns }}

          Removed columns:
          {{ outputs.compare_schemas.vars.removed_columns }}

          Type changes:
          {{ outputs.compare_schemas.vars.type_changes }}

          Breaking changes:
          {{ outputs.compare_schemas.vars.breaking_changes }}

          The incoming dataset can continue
          to downstream processing.

      - id: compatibility_summary
        type: io.kestra.plugin.core.output.OutputValues

        description: >
          Publishes the successful schema compatibility decision.

        values:
          status: "COMPATIBLE"
          message: "Incoming dataset schema is compatible with the expected contract."
          added_column_count: "{{ outputs.compare_schemas.vars.added_columns | length }}"
          removed_column_count: "{{ outputs.compare_schemas.vars.removed_columns | length }}"
          type_change_count: "{{ outputs.compare_schemas.vars.type_changes | length }}"
          breaking_change_count: "{{ outputs.compare_schemas.vars.breaking_changes | length }}"

    else:

      # -----------------------------------------------------
      # FAILURE PATH
      # -----------------------------------------------------

      - id: breaking_schema_log
        type: io.kestra.plugin.core.log.Log

        description: >
          Logs all detected breaking schema changes before stopping downstream
          processing.

        message: |
          ==========================================
          SCHEMA COMPATIBILITY CHECK FAILED
          ==========================================

          Breaking changes detected:
          {{ outputs.compare_schemas.vars.breaking_changes }}

          Added columns:
          {{ outputs.compare_schemas.vars.added_columns }}

          Removed columns:
          {{ outputs.compare_schemas.vars.removed_columns }}

          Type changes:
          {{ outputs.compare_schemas.vars.type_changes }}

          Downstream processing will be blocked.

      - id: fail_incompatible_schema
        type: io.kestra.plugin.core.execution.Fail

        description: >
          Stops the workflow when the incoming dataset schema violates the
          expected schema contract.

        errorMessage: >
          Dataset schema is incompatible with the expected schema contract.
          Review the breaking changes reported by this execution.