Audit a Parquet lake for missing, empty, short and doubled daily partitions

id: parquet-partition-completeness-audit
namespace: company.team

description: >
  Audit a date-partitioned Parquet lake against a calendar by reading only the
  file footers. Report the days that are missing, empty, short or possibly
  loaded twice, the partitions that are not valid days, and the date ranges to
  re-run.

inputs:
  - id: demo_scenario
    type: SELECT
    displayName: Demo scenario
    description: >
      Used only when lake_glob is empty. GAPS: a 90-day lake with 3 missing
      days, 1 empty, 1 short, 1 doubled and 3 stray partitions. CLEAN: a
      complete lake. EMPTY_LAKE: no files at all. WRONG_KEY: the lake is
      partitioned by ds, not dt. COMPACT_DATES: the folders are named like
      dt=20260819, which is not YYYY-MM-DD.
    values:
      - GAPS
      - CLEAN
      - EMPTY_LAKE
      - WRONG_KEY
      - COMPACT_DATES
    defaults: GAPS

  - id: lake_glob
    type: STRING
    required: false
    displayName: Parquet files to audit
    description: >
      Glob for your Parquet files, one folder per day, for example
      /data/events/*/*.parquet or s3://bucket/events/*/*.parquet. Leave empty to
      audit the demo lake that this flow builds itself. It must not contain a
      single quote.
    validator: "^[^']*$"

  - id: partition_key
    type: STRING
    displayName: Partition key
    description: Name of the folder level that holds the date, as in dt=2026-08-19.
      The value must be YYYY-MM-DD.
    defaults: dt
    validator: "^[A-Za-z_][A-Za-z0-9_]*$"

  - id: lookback_days
    type: INT
    displayName: Days to audit
    description: How many days to judge, ending settle_days ago. The demo seeds
      defects up to 29 days back, so use 30 or more to see all of them.
    defaults: 90
    min: 7

  - id: settle_days
    type: INT
    displayName: Settle days
    description: Most recent days that are not judged yet, because their partition
      may still be loading. 1 skips today and judges yesterday.
    defaults: 1
    min: 0

  - id: short_ratio
    type: FLOAT
    displayName: Short below
    description: A day with fewer rows than this share of the median of its previous
      same-weekday days is SHORT.
    defaults: 0.5
    min: 0
    max: 1

  - id: double_ratio
    type: FLOAT
    displayName: Doubled above
    description: A day with more rows than this multiple of that median is DOUBLE,
      which can mean it was loaded twice.
    defaults: 1.8
    min: 1

  - id: on_findings
    type: SELECT
    displayName: On findings
    description: REPORT ends the run SUCCESS and lists the findings. FAIL ends it
      FAILED when there is any finding, so a schedule can alert on it.
    values:
      - REPORT
      - FAIL
    defaults: REPORT

triggers:
  - id: nightly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Disabled. In a real pipeline, set the lake_glob default to your
      lake and run this after the daily load.
    cron: "0 6 * * *"
    disabled: true

tasks:
  - id: workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: Shares one working directory between the tasks, so the demo lake
      written by one task is visible to the next.
    tasks:
      - id: build_demo_lake
        type: io.kestra.plugin.jdbc.duckdb.Queries
        runIf: "{{ (inputs.lake_glob ?? '') == '' }}"
        description: >
          Stand-in for your lake, skipped when lake_glob is set. Writes one
          Parquet folder per day, fixed in time: the audit treats 2026-09-01 as
          today for this lake, so the results are the same every day.
        sql: |
          -- One row per calendar day: the audited days plus 28 days of history for the baseline.
          CREATE TABLE days AS
          SELECT DATE '2026-09-01' - CAST(k AS INTEGER) AS dt, k AS ago
          FROM range(0, {{ inputs.lookback_days }} + {{ inputs.settle_days }} + 29) t(k);

          -- Normal volume: about 1,000 rows on weekdays, 450 on Saturday, 300 on Sunday, with a fixed wobble of -6 to +6 percent.
          CREATE TABLE sized AS
          SELECT dt, ago,
                 CAST((CASE isodow(dt) WHEN 6 THEN 450 WHEN 7 THEN 300 ELSE 1000 + (isodow(dt) - 3) * 20 END)
                      * (94 + (ago * 7) % 13) / 100 AS INTEGER) AS normal_rows
          FROM days;

          -- Rows per day in this scenario. 0 means no folder at all. Days ago 0 is today, which is still loading.
          CREATE TABLE demo_days AS
          SELECT dt, ago,
                 CAST(CASE WHEN '{{ inputs.demo_scenario }}' = 'EMPTY_LAKE' THEN 0
                           WHEN ago = 0 THEN normal_rows * 40 // 100
                           WHEN '{{ inputs.demo_scenario }}' <> 'GAPS' THEN normal_rows
                           WHEN ago IN (4, 5, 29) THEN 0
                           WHEN ago = 13 THEN 1
                           WHEN ago = 21 THEN normal_rows * 30 // 100
                           WHEN ago = 9 THEN normal_rows * 250 // 100
                           ELSE normal_rows END AS BIGINT) AS n
          FROM sized;

          CREATE TABLE ev AS
          SELECT strftime(d.dt, CASE WHEN '{{ inputs.demo_scenario }}' = 'COMPACT_DATES' THEN '%Y%m%d' ELSE '%Y-%m-%d' END) AS dt,
                 CAST(i AS BIGINT) AS id, CAST(i % 50 AS INTEGER) AS k
          FROM demo_days d, range(d.n) r(i);

          -- Three stray partitions: a NULL date as Spark writes it, an epoch default, and a day in the future.
          INSERT INTO ev
          SELECT s.dt, CAST(i AS BIGINT), CAST(i % 50 AS INTEGER)
          FROM (VALUES ('__HIVE_DEFAULT_PARTITION__'), ('1970-01-01'), ('2026-09-04')) s(dt), range(5) r(i)
          WHERE '{{ inputs.demo_scenario }}' = 'GAPS';

          COPY (SELECT * FROM ev WHERE '{{ inputs.demo_scenario }}' <> 'WRONG_KEY')
          TO '{{ workingDir }}/lake' (FORMAT parquet, PARTITION_BY (dt), FILENAME_PATTERN 'part-{i}', OVERWRITE_OR_IGNORE);

          COPY (SELECT dt AS ds, id, k FROM ev WHERE '{{ inputs.demo_scenario }}' = 'WRONG_KEY')
          TO '{{ workingDir }}/lake' (FORMAT parquet, PARTITION_BY (ds), FILENAME_PATTERN 'part-{i}', OVERWRITE_OR_IGNORE);

      - id: add_empty_partition
        type: io.kestra.plugin.jdbc.duckdb.Queries
        runIf: "{{ (inputs.lake_glob ?? '') == '' and inputs.demo_scenario == 'GAPS' }}"
        description: >
          Demo only. PARTITION_BY cannot write a file with zero rows, so this
          overwrites the file of 2026-08-19 (13 days before the demo date) with
          an empty one. That is the empty partition of the GAPS scenario.
        sql: |
          COPY (SELECT CAST(0 AS BIGINT) AS id, CAST(0 AS INTEGER) AS k WHERE false)
          TO '{{ workingDir }}/lake/dt=2026-08-19/part-0.parquet' (FORMAT parquet);

      - id: find_files
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: >
          List the files lake_glob matches and check that the folder level named
          by partition_key exists in their paths and holds dates. A wrong path,
          a wrong key or another date format would otherwise read as a lake with
          every day missing.
        fetchType: FETCH_ONE
        sql: |
          CREATE TABLE files AS
          SELECT file FROM glob(COALESCE(NULLIF('{{ inputs.lake_glob ?? '' }}', ''), '{{ workingDir }}/lake/*/*.parquet'));

          WITH c AS (
            SELECT count(*) AS files_total,
                   count(*) FILTER (WHERE regexp_matches(file, '/{{ inputs.partition_key }}=[^/]*/')) AS files_with_key,
                   count(*) FILTER (WHERE try_cast(regexp_extract(file, '/{{ inputs.partition_key }}=([^/]*)/', 1) AS DATE) IS NOT NULL) AS files_with_date
            FROM files
          )
          SELECT files_total, files_with_key, files_with_date,
                 CASE WHEN files_total = 0 THEN 'NO_FILES'
                      WHEN files_with_key = 0 THEN 'NO_PARTITION_KEY'
                      WHEN files_with_date = 0 THEN 'NO_DATES'
                      ELSE 'OK' END AS scan_status,
                 CASE WHEN files_total = 0 THEN 'lake_glob matched no Parquet files. A wrong path would otherwise look like a lake with every day missing. Nothing was audited.'
                      WHEN files_with_key = 0 THEN files_total || ' Parquet files matched, but none has a {{ inputs.partition_key }}= folder in its path. Check partition_key. Nothing was audited.'
                      WHEN files_with_date = 0 THEN files_with_key || ' Parquet files have a {{ inputs.partition_key }}= folder, but none of its values is a date such as 2026-08-19. Nothing was audited.'
                      ELSE files_total || ' Parquet files matched.' END AS reason
          FROM c;

      - id: scan_guard
        type: io.kestra.plugin.core.execution.Fail
        runIf: "{{ (outputs.find_files.outputs | last).row.scan_status != 'OK' }}"
        description: Stop before the audit when there is nothing valid to audit.
        errorMessage: "BLOCKED_{{ (outputs.find_files.outputs | last).row.scan_status
          }}: {{ (outputs.find_files.outputs | last).row.reason }}"

      - id: audit
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: >
          Read the row count of every file from its Parquet footer, group it by
          day, compare each day with the median of the previous four
          same-weekday days, and list the days that are missing, empty, short or
          doubled, the stray partitions, and the date ranges to re-run. It reads
          no data and changes nothing.
        fetchType: FETCH_ONE
        outputFiles:
          - findings
          - ranges
        sql: |
          -- Today is the demo date for the demo lake and the real date otherwise.
          CREATE TABLE ctx AS
          SELECT CASE WHEN COALESCE('{{ inputs.lake_glob ?? '' }}', '') = '' THEN DATE '2026-09-01' ELSE current_date END AS today;

          -- The audited window ends settle_days ago and is lookback_days long.
          CREATE TABLE win AS
          SELECT today - {{ inputs.settle_days }} AS d_to,
                 today - {{ inputs.settle_days }} - ({{ inputs.lookback_days }} - 1) AS d_from
          FROM ctx;

          -- Footer scan: one row per file with its row count. No data is read.
          CREATE TABLE meta AS
          SELECT file_name, num_rows
          FROM parquet_file_metadata(COALESCE(NULLIF('{{ inputs.lake_glob ?? '' }}', ''), '{{ workingDir }}/lake/*/*.parquet'));

          -- One row per partition folder. part is '' for a file that has no partition key in its path.
          CREATE TABLE inv AS
          SELECT regexp_extract(file_name, '/{{ inputs.partition_key }}=([^/]*)/', 1) AS part,
                 CAST(sum(num_rows) AS BIGINT) AS n_rows,
                 count(*) AS n_files
          FROM meta
          GROUP BY 1;

          -- Judge every day of the window. The 28 days before it only feed the baseline.
          CREATE TABLE judged AS
          WITH cal AS (
            SELECT CAST(g AS DATE) AS dt
            FROM win, generate_series(d_from - 28, d_to, INTERVAL 1 DAY) t(g)
          ), by_day AS (
            -- 2026-8-5 and 2026-08-05 are two folders but one day.
            SELECT try_cast(part AS DATE) AS dt, CAST(sum(n_rows) AS BIGINT) AS n_rows, CAST(sum(n_files) AS BIGINT) AS n_files
            FROM inv
            WHERE try_cast(part AS DATE) IS NOT NULL
            GROUP BY 1
          ), joined AS (
            SELECT c.dt, b.n_rows, b.n_files
            FROM cal c LEFT JOIN by_day b ON b.dt = c.dt
          ), based AS (
            SELECT *, median(n_rows) OVER w AS baseline, count(n_rows) OVER w AS base_points
            FROM joined
            WINDOW w AS (PARTITION BY isodow(dt) ORDER BY dt ROWS BETWEEN 4 PRECEDING AND 1 PRECEDING)
          )
          SELECT dt, n_rows, n_files, baseline, base_points,
                 CASE WHEN n_files IS NULL THEN 'MISSING'
                      WHEN n_rows = 0 THEN 'EMPTY'
                      WHEN base_points < 3 THEN 'NO_BASELINE'
                      WHEN n_rows < {{ inputs.short_ratio }} * baseline THEN 'SHORT'
                      WHEN n_rows > {{ inputs.double_ratio }} * baseline THEN 'DOUBLE'
                      ELSE 'OK' END AS status
          FROM based, win
          WHERE dt >= d_from;

          -- Partitions that are not a day of the calendar.
          CREATE TABLE stray AS
          SELECT CASE WHEN part = '' THEN '(no {{ inputs.partition_key }}= in the path)' ELSE part END AS part,
                 n_rows, n_files,
                 CASE WHEN part = '' THEN 'no partition key in the path'
                      WHEN try_cast(part AS DATE) IS NULL THEN 'not a date'
                      WHEN try_cast(part AS DATE) > today + 1 THEN 'more than a day in the future'
                      ELSE 'before 1971, usually a NULL timestamp cast to 0' END AS reason
          FROM inv, ctx
          WHERE part = ''
             OR try_cast(part AS DATE) IS NULL
             OR try_cast(part AS DATE) > today + 1
             OR try_cast(part AS DATE) < DATE '1971-01-01';

          CREATE TABLE findings AS
          SELECT CAST(dt AS VARCHAR) AS partition_value, status,
                 coalesce(n_rows, 0) AS row_count,
                 CAST(round(baseline) AS BIGINT) AS baseline_rows,
                 CASE WHEN baseline > 0 AND n_rows IS NOT NULL THEN round(n_rows / baseline, 2) END AS ratio,
                 coalesce(n_files, 0) AS files,
                 CASE status
                   WHEN 'MISSING' THEN 'no Parquet file for this day'
                   WHEN 'EMPTY' THEN 'files exist but hold 0 rows'
                   WHEN 'SHORT' THEN 'fewer rows than {{ inputs.short_ratio }} times the same-weekday median'
                   ELSE 'more rows than {{ inputs.double_ratio }} times the same-weekday median, possibly loaded twice' END AS note
          FROM judged
          WHERE status IN ('MISSING', 'EMPTY', 'SHORT', 'DOUBLE')
          UNION ALL
          SELECT part, 'STRAY', n_rows, NULL, NULL, n_files, reason FROM stray
          ORDER BY 1;

          -- Contiguous flagged days become one range each: a day minus its position in the list is constant inside a run.
          CREATE TABLE ranges AS
          WITH f AS (
            SELECT dt, status, dt - CAST(row_number() OVER (ORDER BY dt) AS INTEGER) AS grp
            FROM judged
            WHERE status IN ('MISSING', 'EMPTY', 'SHORT', 'DOUBLE')
          )
          SELECT CAST(min(dt) AS VARCHAR) AS start_date,
                 CAST(max(dt) AS VARCHAR) AS end_date,
                 count(*) AS days,
                 string_agg(DISTINCT status, ', ' ORDER BY status) AS statuses
          FROM f
          GROUP BY grp
          ORDER BY min(dt);

          COPY findings TO '{{ outputFiles.findings }}' (HEADER, DELIMITER ',');
          COPY ranges TO '{{ outputFiles.ranges }}' (HEADER, DELIMITER ',');

          WITH s AS (
            SELECT (SELECT count(*) FROM meta) AS files,
                   (SELECT count(*) FROM judged) AS days_audited,
                   (SELECT count(*) FROM judged WHERE status = 'OK') AS ok_days,
                   (SELECT count(*) FROM judged WHERE status = 'NO_BASELINE') AS no_baseline_days,
                   (SELECT count(*) FROM judged WHERE status = 'MISSING') AS missing_days,
                   (SELECT count(*) FROM judged WHERE status = 'EMPTY') AS empty_days,
                   (SELECT count(*) FROM judged WHERE status = 'SHORT') AS short_days,
                   (SELECT count(*) FROM judged WHERE status = 'DOUBLE') AS double_days,
                   (SELECT count(*) FROM stray) AS stray_partitions,
                   (SELECT count(*) FROM ranges) AS ranges,
                   (SELECT CAST(d_from AS VARCHAR) FROM win) AS audited_from,
                   (SELECT CAST(d_to AS VARCHAR) FROM win) AS audited_to,
                   (SELECT coalesce(string_agg(start_date || CASE WHEN end_date <> start_date THEN '..' || end_date ELSE '' END, ', ' ORDER BY start_date), '') FROM ranges) AS plan
          ), d AS (
            SELECT *,
                   CASE WHEN missing_days + empty_days + short_days + double_days + stray_partitions > 0 THEN 'FINDINGS' ELSE 'COMPLETE' END AS decision
            FROM s
          )
          SELECT *,
                 CASE decision
                   WHEN 'FINDINGS' THEN 'Audited ' || audited_from || ' to ' || audited_to || ' (' || days_audited || ' days): '
                                        || missing_days || ' missing, ' || empty_days || ' empty, ' || short_days || ' short, '
                                        || double_days || ' possibly loaded twice, ' || stray_partitions || ' stray partition(s).'
                   ELSE 'Audited ' || audited_from || ' to ' || audited_to || ' (' || days_audited || ' days): every day has data in line with its weekday baseline and there are no stray partitions.'
                 END
                 || CASE WHEN no_baseline_days > 0 THEN ' ' || no_baseline_days || ' day(s) had too little history for a volume check.' ELSE '' END AS reason
          FROM d;

  - id: result
    type: io.kestra.plugin.core.output.OutputValues
    values:
      row: "{{ (outputs.audit.outputs | last).row | toJson }}"

  - id: log_audit
    type: io.kestra.plugin.core.log.Log
    message: >-
      {{ outputs.result.values.row | jq('.reason') | first }} Files read: {{
      outputs.result.values.row | jq('.files') | first }}. Missing: {{
      outputs.result.values.row | jq('.missing_days') | first }}, empty: {{
      outputs.result.values.row | jq('.empty_days') | first }}, short: {{
      outputs.result.values.row | jq('.short_days') | first }}, doubled: {{
      outputs.result.values.row | jq('.double_days') | first }}, stray: {{
      outputs.result.values.row | jq('.stray_partitions') | first }}. Ranges to
      re-run: {{ outputs.result.values.row | jq('.plan') | first }}

  - id: route
    type: io.kestra.plugin.core.flow.Switch
    description: COMPLETE logs and ends. FINDINGS lists the ranges and, when
      on_findings is FAIL, fails the run.
    value: "{{ outputs.result.values.row | jq('.decision') | first }}"
    cases:
      COMPLETE:
        - id: log_complete
          type: io.kestra.plugin.core.log.Log
          message: "The lake is complete for the audited window."

      FINDINGS:
        - id: log_findings
          type: io.kestra.plugin.core.log.Log
          level: WARN
          message: >-
            {{ outputs.result.values.row | jq('.ranges') | first }} range(s) to
            re-run, by day: {{ outputs.result.values.row | jq('.plan') | first
            }}. The details are in outputs.audit.outputFiles.findings and
            outputs.audit.outputFiles.ranges.

        - id: fail_on_findings
          type: io.kestra.plugin.core.execution.Fail
          runIf: "{{ inputs.on_findings == 'FAIL' }}"
          errorMessage: "FINDINGS: {{ outputs.result.values.row | jq('.reason') | first }}"

    defaults:
      - id: unexpected_decision
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Unexpected decision {{ outputs.result.values.row |
          jq('.decision') | first }}. Check the audit task logs."

errors:
  - id: report_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: "Partition audit failed in {{ flow.id }}, execution {{ execution.id }}.
      A BLOCKED_ message from scan_guard means the lake_glob or partition_key
      does not match any partition folder, which is the expected result for the
      EMPTY_LAKE, WRONG_KEY and COMPACT_DATES demo scenarios. A FINDINGS message
      means on_findings is FAIL. Otherwise read the audit task logs: an
      unreadable Parquet file stops the scan and its path is in the error."