Parquet Data Quality Contract Gate with DuckDB

id: duckdb-parquet-data-quality-contract-gate
namespace: company.team
description: |
  Validate a Parquet dataset against a comprehensive data contract using embedded DuckDB.
  Enforces required columns and data types, detects unexpected columns, calculates null ratios,
  asserts primary-key uniqueness, and validates numeric value ranges before downstream loading.

inputs:
  - id: parquet_uri
    type: STRING
    displayName: Parquet File URI
    description: Kestra internal storage URI (kestra://...) or path to the target
      Parquet file to audit. When empty, a self-contained sample dataset is
      generated in-memory.
    defaults: ""

  - id: inject_bad_data
    type: BOOL
    displayName: Inject Contract Violations
    description: When true and parquet_uri is empty, the generated sample contains
      violations (null primary key, duplicate key, out-of-range value,
      unexpected column) to test gate failure.
    defaults: false

  - id: max_null_ratio
    type: FLOAT
    displayName: Maximum Allowable Null Ratio
    description: Threshold for allowable null values per required column (e.g. 0.01
      allows up to 1% nulls).
    defaults: 0.01

  - id: min_amount
    type: FLOAT
    displayName: Minimum Valid Amount
    description: Lower bound for the transaction amount numeric column contract check.
    defaults: 0.01

  - id: max_amount
    type: FLOAT
    displayName: Maximum Valid Amount
    description: Upper bound for the transaction amount numeric column contract check.
    defaults: 100000.0

  - id: fail_on_violation
    type: BOOL
    displayName: Fail on Violation
    description: Whether to abort the pipeline with an execution error when one or
      more contract checks fail.
    defaults: true

triggers:
  - id: daily_contract_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled daily contract audit at 06:00 UTC. Shipped disabled so
      manual and demo runs can be executed on demand.
    cron: "0 6 * * *"
    disabled: true

tasks:
  - id: generate_sample_data
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Generates a self-contained transactions Parquet file for demo runs
      when no external parquet_uri is provided. Exists solely to ensure the
      blueprint runs out-of-the-box as demo.
    timeout: PT2M
    outputFiles:
      - sample.parquet
    sql: |
      {% if inputs.inject_bad_data %}
      -- Poisoned dataset with deliberate contract violations:
      -- 1. NULL primary key on row 1
      -- 2. Duplicate order_id (3) on row 2
      -- 3. Out-of-range negative amount (-25.50) on row 3
      -- 4. Unexpected extra column ('unauthorized_tracking_token')
      CREATE TABLE orders AS
      SELECT
        CASE WHEN i = 1 THEN NULL WHEN i = 2 THEN 3 ELSE i END AS order_id,
        100 + i AS customer_id,
        CASE WHEN i = 3 THEN -25.50 ELSE ROUND(10.0 + (i * 5.25), 2) END AS amount,
        'COMPLETED' AS status,
        DATE '2025-01-01' + i AS order_date,
        'unauthorized_tracking_token' AS unexpected_column
      FROM range(1, 101) t(i);
      {% else %}
      -- Compliant dataset satisfying all schema and data contract constraints
      CREATE TABLE orders AS
      SELECT
        i AS order_id,
        100 + i AS customer_id,
        ROUND(10.0 + (i * 5.25), 2) AS amount,
        'COMPLETED' AS status,
        DATE '2025-01-01' + i AS order_date
      FROM range(1, 101) t(i);
      {% endif %}

      COPY orders TO '{{ outputFiles["sample.parquet"] }}' (FORMAT PARQUET);

  - id: validate_contract
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Run comprehensive SQL contract checks against the target Parquet
      file and return structured check results.
    timeout: PT2M
    inputFiles:
      target.parquet: "{{ inputs.parquet_uri is not empty ? inputs.parquet_uri :
        outputs.generate_sample_data.outputFiles['sample.parquet'] }}"
    fetchType: FETCH
    sql: |
      WITH data AS (
        SELECT * FROM read_parquet('target.parquet')
      ),
      cols AS (
        SELECT column_name, column_type FROM (DESCRIBE SELECT * FROM read_parquet('target.parquet'))
      ),
      checks AS (
        -- Check 1: Row count greater than zero
        SELECT
          'ROW_COUNT_GREATER_THAN_ZERO' AS check_name,
          (COUNT(*) > 0) AS passed,
          'Observed rows: ' || COUNT(*)::VARCHAR AS details
        FROM data

        UNION ALL

        -- Check 2: Required columns exist with expected data types
        SELECT
          'REQUIRED_COLUMNS_AND_TYPES' AS check_name,
          (
            COUNT(CASE WHEN column_name = 'order_id' AND column_type IN ('BIGINT', 'INTEGER', 'INT8', 'INT4') THEN 1 END) = 1 AND
            COUNT(CASE WHEN column_name = 'customer_id' AND column_type IN ('BIGINT', 'INTEGER', 'INT8', 'INT4') THEN 1 END) = 1 AND
            COUNT(CASE WHEN column_name = 'amount' AND (column_type LIKE 'DECIMAL%' OR column_type IN ('DOUBLE','FLOAT')) THEN 1 END) = 1 AND
            COUNT(CASE WHEN column_name = 'status' AND column_type LIKE '%VARCHAR%' THEN 1 END) = 1 AND
            COUNT(CASE WHEN column_name = 'order_date' AND column_type = 'DATE' THEN 1 END) = 1
          ) AS passed,
          'Required columns checked: order_id, customer_id, amount, status, order_date' AS details
        FROM cols

        UNION ALL

        -- Check 3: No unexpected extra columns present
        SELECT
          'NO_UNEXPECTED_COLUMNS' AS check_name,
          (COUNT(CASE WHEN column_name NOT IN ('order_id', 'customer_id', 'amount', 'status', 'order_date') THEN 1 END) = 0) AS passed,
          COALESCE(
            'Unexpected columns found: ' || string_agg(CASE WHEN column_name NOT IN ('order_id', 'customer_id', 'amount', 'status', 'order_date') THEN column_name END, ', '),
            'Only expected contract columns present'
          ) AS details
        FROM cols

        UNION ALL

        -- Check 4: Null ratio per required column is at or below max_null_ratio
        SELECT
          'NULL_RATIO_BELOW_THRESHOLD' AS check_name,
          (
            (COUNT(CASE WHEN order_id IS NULL THEN 1 END)::DOUBLE / GREATEST(COUNT(*), 1)) <= {{ inputs.max_null_ratio }} AND
            (COUNT(CASE WHEN customer_id IS NULL THEN 1 END)::DOUBLE / GREATEST(COUNT(*), 1)) <= {{ inputs.max_null_ratio }} AND
            (COUNT(CASE WHEN amount IS NULL THEN 1 END)::DOUBLE / GREATEST(COUNT(*), 1)) <= {{ inputs.max_null_ratio }}
          ) AS passed,
          'Null ratios: order_id=' || ROUND(COUNT(CASE WHEN order_id IS NULL THEN 1 END)::DOUBLE / GREATEST(COUNT(*), 1), 4)::VARCHAR ||
          ', customer_id=' || ROUND(COUNT(CASE WHEN customer_id IS NULL THEN 1 END)::DOUBLE / GREATEST(COUNT(*), 1), 4)::VARCHAR ||
          ', amount=' || ROUND(COUNT(CASE WHEN amount IS NULL THEN 1 END)::DOUBLE / GREATEST(COUNT(*), 1), 4)::VARCHAR ||
          ' (threshold: ' || {{ inputs.max_null_ratio }}::VARCHAR || ')' AS details
        FROM data

        UNION ALL

        -- Check 5: Primary key uniqueness
        SELECT
          'PRIMARY_KEY_UNIQUENESS' AS check_name,
          (COUNT(*) = COUNT(DISTINCT order_id)) AS passed,
          'Rows: ' || COUNT(*)::VARCHAR || ', null keys: ' || COUNT(CASE WHEN order_id IS NULL THEN 1 END)::VARCHAR || ', distinct keys: ' || COUNT(DISTINCT order_id)::VARCHAR AS details
        FROM data

        UNION ALL

        -- Check 6: Value range on numeric amount column
        SELECT
          'NUMERIC_VALUE_RANGE' AS check_name,
          (
            COUNT(CASE WHEN amount < {{ inputs.min_amount }} OR amount > {{ inputs.max_amount }} THEN 1 END) = 0
          ) AS passed,
          'Observed range: [' || MIN(amount)::VARCHAR || ', ' || MAX(amount)::VARCHAR ||
          '], allowed: [' || {{ inputs.min_amount }}::VARCHAR || ', ' || {{ inputs.max_amount }}::VARCHAR || ']' AS details
        FROM data
      )
      SELECT * FROM checks ORDER BY check_name;

  - id: log_contract_summary
    type: io.kestra.plugin.core.log.Log
    description: Print a structured pass/fail report for all evaluated contract
      checks to execution logs.
    message: |
      === Parquet Data Quality Contract Summary ===
      {% for check in outputs.validate_contract.rows %}
      [{{ check.passed ? 'PASS' : 'FAIL' }}] {{ check.check_name }}: {{ check.details }}
      {% endfor %}

  - id: contract_gate
    type: io.kestra.plugin.core.flow.If
    description: Evaluate contract results and abort pipeline execution if any rule
      failed and fail_on_violation is enabled.
    condition: "{{ inputs.fail_on_violation and (outputs.validate_contract.rows |
      jq('[.[] | select(.passed == false)] | length') | first > 0) }}"
    then:
      - id: fail_pipeline
        type: io.kestra.plugin.core.execution.Fail
        description: Halt pipeline with details of all violated contract checks.
        errorMessage: "Parquet data quality contract violated! Failed checks: {{
          outputs.validate_contract.rows | jq('[.[] | select(.passed == false) |
          .check_name + \" (\" + .details + \")\"] | join(\"; \")') | first }}"