Parquet Data Skipping Audit with DuckDB

id: parquet-data-skipping-audit
namespace: company.team

description: >
  Audit how well a Parquet dataset can skip data for one filter column, from the
  min/max statistics in its file footers. It counts how many row groups and
  files a lookup has to read, flags a layout that is too scattered, and can
  write a sorted copy and prove that the copy has the same rows and reads less.

inputs:
  - id: demo_scenario
    type: SELECT
    displayName: Demo scenario
    description: >
      Used only when lake_glob is empty. The demo lake has 20 files of 4 row
      groups, with the same rows in every scenario. SCATTERED: written in
      arrival order, so event_time is clustered and customer_id is not.
      CLUSTERED: sorted by customer_id. LOCALLY_SORTED: every file is sorted by
      customer_id, but each file covers the whole range.
    values:
      - SCATTERED
      - CLUSTERED
      - LOCALLY_SORTED
    defaults: SCATTERED

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

  - id: column
    type: STRING
    displayName: Column to audit
    description: >
      The column your queries filter on. It must be a top-level column stored in
      the files, not a partition folder. Integers, FLOAT, DOUBLE, DECIMAL, DATE,
      TIMESTAMP (with or without time zone) and VARCHAR are supported. The demo
      lake also has event_time, customer_ref, source (one value), notes (all
      NULL) and tags (a list), to show the guards.
    defaults: customer_id
    validator: "^[A-Za-z_][A-Za-z0-9_]*$"

  - id: max_read_pct
    type: INT
    displayName: Most row groups to read (percent)
    description: >
      The share of row groups that a lookup on the column may read. Above it the
      layout is SKIPPING_POOR. A column with few distinct values cannot get
      below 100 divided by that number, so raise it for such a column.
    defaults: 10
    min: 1
    max: 100

  - id: mode
    type: SELECT
    displayName: Mode
    description: >
      AUDIT only measures. REWRITE_COPY also writes the rows sorted by the
      column to a new folder, checks the copy and measures it, when the layout
      is poor. Your files are never changed.
    values:
      - AUDIT
      - REWRITE_COPY
    defaults: AUDIT

  - id: rewrite_to
    type: STRING
    required: false
    displayName: Folder for the sorted copy
    description: >
      Only for REWRITE_COPY. A folder that does not exist or is empty, outside
      the files you audit, and without a single quote or a semicolon. Empty uses
      a folder in the working directory of the demo lake. It is required when
      lake_glob is set.
    validator: "^[^';]*$"

  - id: on_poor_skipping
    type: SELECT
    displayName: On poor skipping
    description: >
      REPORT ends the run SUCCESS and says what it found. FAIL ends it FAILED
      when the layout is poor, or when the sorted copy does not reach the
      target, so a schedule can alert on it.
    values:
      - REPORT
      - FAIL
    defaults: REPORT

triggers:
  - id: weekly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Disabled. In a real pipeline, set the lake_glob and column defaults
      to your dataset first.
    cron: "0 6 * * 1"
    disabled: true

tasks:
  - id: workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: Shares one working directory between the tasks, so the demo lake
      and the sorted copy are visible to every task.
    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 490,000
          rows as 20 Parquet files of 4 row groups. The rows are the same in
          every scenario and only their order changes. One thread keeps the
          write order, so the files and row groups are the same on every run.
        sql: |
          SET threads = 1;

          -- Rows in arrival order. The key is spread over 100,003 values and event_time only grows.
          CREATE TABLE src AS
          SELECT CAST(i AS BIGINT) AS id,
                 TIMESTAMP '2026-09-01 00:00:00' + to_seconds(CAST(i AS BIGINT) * 5) AS event_time,
                 CAST(100000 + (i * 7919) % 100003 AS BIGINT) AS customer_id,
                 'C-' || lpad(CAST(100000 + (i * 7919) % 100003 AS VARCHAR), 7, '0') AS customer_ref,
                 'web' AS source,
                 CAST(NULL AS VARCHAR) AS notes,
                 [CAST(i % 3 AS VARCHAR)] AS tags
          FROM range(490000) t(i);

          -- The scenario is only the order the rows are written in. A file takes 24,576 rows: 4 row groups of 6,144.
          COPY (
            SELECT id, event_time, customer_id, customer_ref, source, notes, tags
            FROM src
            ORDER BY CASE WHEN '{{ inputs.demo_scenario }}' = 'LOCALLY_SORTED' THEN id // 24576 ELSE 0 END,
                     CASE WHEN '{{ inputs.demo_scenario }}' = 'SCATTERED' THEN 0 ELSE customer_id END,
                     id
          ) TO '{{ workingDir }}/lake' (FORMAT parquet, ROW_GROUP_SIZE 6144, ROW_GROUPS_PER_FILE 4);

      - id: find_files
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: >
          List the files lake_glob matches and, for REWRITE_COPY, check the
          folder for the copy. A wrong path would otherwise look like an empty
          lake, and a copy written into the lake would be audited with it.
        fetchType: FETCH_ONE
        sql: |
          CREATE TABLE files AS
          SELECT file FROM glob(COALESCE(NULLIF('{{ inputs.lake_glob ?? '' }}', ''), '{{ workingDir }}/lake/*.parquet'));

          CREATE TABLE target AS
          SELECT COALESCE(NULLIF(rtrim('{{ inputs.rewrite_to ?? '' }}', '/'), ''), '{{ workingDir }}/sorted') AS dir,
                 NULLIF('{{ inputs.rewrite_to ?? '' }}', '') IS NOT NULL AS given,
                 NULLIF('{{ inputs.lake_glob ?? '' }}', '') IS NOT NULL AS own_lake,
                 '{{ inputs.mode }}' = 'REWRITE_COPY' AS rewrite;

          CREATE TABLE in_target AS
          SELECT file FROM glob(COALESCE(NULLIF('{{ inputs.rewrite_to ?? '' }}', ''), '{{ workingDir }}/sorted') || '/**');

          WITH c AS (
            SELECT (SELECT count(*) FROM files) AS files_total,
                   (SELECT count(*) FROM in_target) AS target_files,
                   (SELECT count(*) FROM files f, target t WHERE starts_with(f.file, t.dir || '/')) AS lake_files_in_target,
                   t.dir, t.given, t.own_lake, t.rewrite,
                   COALESCE(NULLIF('{{ inputs.lake_glob ?? '' }}', ''), '{{ workingDir }}/lake/*.parquet') AS lake_glob
            FROM target t
          )
          SELECT files_total, lake_glob, dir AS target_dir,
                 CASE WHEN files_total = 0 THEN 'NO_FILES'
                      WHEN rewrite AND own_lake AND NOT given THEN 'NO_REWRITE_TARGET'
                      WHEN rewrite AND lake_files_in_target > 0 THEN 'OUTPUT_INSIDE_LAKE'
                      WHEN rewrite AND target_files > 0 THEN 'OUTPUT_NOT_EMPTY'
                      ELSE 'OK' END AS scan_status,
                 CASE WHEN files_total = 0 THEN 'lake_glob matched no files. A wrong path would otherwise look like an empty lake. Nothing was audited.'
                      WHEN rewrite AND own_lake AND NOT given THEN 'mode is REWRITE_COPY and lake_glob is set, so rewrite_to is required. The copy is never written next to your files by default. Nothing was changed.'
                      WHEN rewrite AND lake_files_in_target > 0 THEN 'rewrite_to contains ' || lake_files_in_target || ' of the files lake_glob matches. The copy must be outside the audited files. Nothing was changed.'
                      WHEN rewrite AND target_files > 0 THEN 'rewrite_to already holds ' || target_files || ' file(s). Use a folder that does not exist or is empty. Nothing was changed.'
                      ELSE files_total || ' 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 to audit or the copy
          target is unsafe.
        errorMessage: "BLOCKED_{{ (outputs.find_files.outputs | last).row.scan_status
          }}: {{ (outputs.find_files.outputs | last).row.reason }}"

      - id: inspect_column
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: >
          Check that the column exists in the files, that its type is one the
          audit can order, that the footers hold min/max statistics for it, and
          that it holds more than one value. Otherwise the audit would measure
          nothing.
        fetchType: FETCH_ONE
        sql: |
          CREATE TABLE cols AS
          SELECT column_name, column_type
          FROM (DESCRIBE SELECT * FROM read_parquet('{{ (outputs.find_files.outputs | last).row.lake_glob }}', hive_partitioning = false, union_by_name = true))
          WHERE lower(column_name) = lower('{{ inputs.column }}');

          CREATE TABLE fm AS
          SELECT stats_min_value AS lo, stats_max_value AS hi
          FROM parquet_metadata('{{ (outputs.find_files.outputs | last).row.lake_glob }}')
          WHERE lower(path_in_schema) = lower('{{ inputs.column }}');

          WITH c AS (
            SELECT (SELECT count(*) FROM cols) AS found,
                   (SELECT any_value(column_type) FROM cols) AS column_type,
                   (SELECT regexp_matches(any_value(column_type), '^(U?(TINYINT|SMALLINT|INTEGER|BIGINT)|HUGEINT|FLOAT|DOUBLE|DECIMAL\(\d+,\d+\)|DATE|TIMESTAMP( WITH TIME ZONE)?|TIMESTAMP_NS|VARCHAR)$') FROM cols) AS type_ok,
                   (SELECT count(*) FROM fm) AS row_groups,
                   (SELECT count(*) FROM fm WHERE lo IS NOT NULL AND hi IS NOT NULL) AS with_stats,
                   (SELECT count(DISTINCT v) FROM (SELECT lo AS v FROM fm UNION ALL SELECT hi FROM fm) WHERE v IS NOT NULL) AS distinct_bounds
          )
          SELECT column_type, row_groups, with_stats,
                 CASE WHEN found = 0 THEN 'COLUMN_NOT_FOUND'
                      WHEN NOT type_ok THEN 'UNSUPPORTED_TYPE'
                      WHEN with_stats = 0 THEN 'NO_STATISTICS'
                      WHEN distinct_bounds = 1 THEN 'CONSTANT_COLUMN'
                      ELSE 'OK' END AS column_status,
                 CASE WHEN found = 0 THEN '{{ inputs.column }} is not a top-level column stored in the files. A partition folder such as dt=2026-08-19 is not stored in them, so there is nothing to audit. Nothing was audited.'
                      WHEN NOT type_ok THEN '{{ inputs.column }} has type ' || column_type || '. The audit orders integers, FLOAT, DOUBLE, DECIMAL, DATE, TIMESTAMP (with or without time zone) and VARCHAR. Nothing was audited.'
                      WHEN with_stats = 0 THEN 'None of the ' || row_groups || ' row groups has min/max statistics for {{ inputs.column }}. It is all NULL, or the files were written without statistics, and then no engine can skip anything. Nothing was audited.'
                      WHEN distinct_bounds = 1 THEN '{{ inputs.column }} holds a single value in every row group, so no layout can skip anything. Nothing was audited.'
                      ELSE '{{ inputs.column }} is ' || column_type || ' with statistics in ' || with_stats || ' of ' || row_groups || ' row groups.' END AS reason
          FROM c;

      - id: column_guard
        type: io.kestra.plugin.core.execution.Fail
        runIf: "{{ (outputs.inspect_column.outputs | last).row.column_status != 'OK' }}"
        description: Stop before the audit when the column cannot be measured.
        errorMessage: "BLOCKED_{{ (outputs.inspect_column.outputs |
          last).row.column_status }}: {{ (outputs.inspect_column.outputs |
          last).row.reason }}"

      - id: audit
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: >
          Read the min/max range of every row group from the footers, count for
          each range how many other ranges overlap it, and turn that into the
          share of row groups and files that a lookup has to read. It reads no
          data and changes nothing. The measurement is repeated in verify_copy
          for the sorted copy, keep the two in step.
        fetchType: FETCH_ONE
        outputFiles:
          - widest_files
        sql: |
          -- Footer scan: one row per row group for the column. No data is read.
          CREATE TABLE rg AS
          SELECT file_name AS f, row_group_id AS g, row_group_num_rows AS n, num_values, stats_null_count AS nulls,
                 TRY_CAST(stats_min_value AS {{ (outputs.inspect_column.outputs | last).row.column_type }}) AS lo,
                 TRY_CAST(stats_max_value AS {{ (outputs.inspect_column.outputs | last).row.column_type }}) AS hi
          FROM parquet_metadata('{{ (outputs.find_files.outputs | last).row.lake_glob }}')
          WHERE lower(path_in_schema) = lower('{{ inputs.column }}');

          -- Measurement. NO_VALUES is all NULL, a lookup never reads it. NO_STATS has no usable range, so it is always read.
          CREATE TABLE kinds AS
          SELECT f, g, n, lo, hi,
                 CASE WHEN COALESCE(nulls, -1) = num_values THEN 'NO_VALUES'
                      WHEN lo IS NULL OR hi IS NULL THEN 'NO_STATS'
                      ELSE 'RANGED' END AS kind
          FROM rg;

          -- A file is the span of its row groups. A file with a row group without statistics is left out and always read.
          CREATE TABLE units AS
          SELECT 'row_group' AS grain, f || '#' || g AS id, lo, hi FROM kinds WHERE kind = 'RANGED'
          UNION ALL
          SELECT 'file', f, min(lo), max(hi) FROM kinds
          WHERE kind = 'RANGED' AND f NOT IN (SELECT f FROM kinds WHERE kind = 'NO_STATS')
          GROUP BY f;

          -- A range overlaps mine when it starts at or before my end and ends at or after my start. Another range
          -- does not overlap when it starts after my end or ends before my start, never both. So the overlaps are
          -- the ranges starting at or before my end, minus the ranges ending before my start, minus mine.
          -- Pass A counts starts up to each end. Starts sort before the question, so a tie counts as overlap.
          -- Pass B counts ends before each start. Questions sort before the ends, so a tie counts as overlap.
          -- Running sums over sorted values, no join: a join of these tables is planned badly in a transaction.
          CREATE TABLE ov AS
          WITH s AS (
            SELECT grain, 'A' AS pass, id, lo AS v, 0 AS ord, 1 AS step, 0 AS ask, lo AS u_lo, hi AS u_hi FROM units
            UNION ALL
            SELECT grain, 'A', id, hi, 1, 0, 1, lo, hi FROM units
            UNION ALL
            SELECT grain, 'B', id, lo, 0, 0, 1, lo, hi FROM units
            UNION ALL
            SELECT grain, 'B', id, hi, 1, 1, 0, lo, hi FROM units
          ), c AS (
            SELECT grain, pass, id, ask, u_lo, u_hi,
                   sum(step) OVER (PARTITION BY grain, pass ORDER BY v, ord ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running
            FROM s
          )
          SELECT grain, id, min(u_lo) AS lo, min(u_hi) AS hi,
                 CAST(max(CASE WHEN pass = 'A' THEN running END) - max(CASE WHEN pass = 'B' THEN running END) - 1 AS BIGINT) AS n_overlap
          FROM c
          WHERE ask = 1
          GROUP BY grain, id;

          CREATE TABLE r AS
          WITH m AS (
            SELECT (SELECT count(*) FROM kinds WHERE kind <> 'NO_VALUES') AS row_groups,
                   (SELECT count(DISTINCT f) FROM kinds WHERE kind <> 'NO_VALUES') AS files,
                   (SELECT count(*) FROM kinds WHERE kind = 'NO_STATS') AS rg_no_stats,
                   (SELECT count(DISTINCT f) FROM kinds WHERE kind = 'NO_STATS') AS files_no_stats,
                   (SELECT count(*) FROM kinds WHERE kind = 'NO_VALUES') AS rg_no_values,
                   (SELECT avg(n_overlap) FROM ov WHERE grain = 'row_group') AS rg_overlaps,
                   (SELECT max(n_overlap) FROM ov WHERE grain = 'row_group') AS rg_overlaps_max,
                   (SELECT avg(n_overlap) FROM ov WHERE grain = 'file') AS file_overlaps,
                   (SELECT CAST(median(n) AS BIGINT) FROM kinds) AS row_group_size,
                   (SELECT greatest(1, CAST(median(c) AS BIGINT)) FROM (SELECT count(*) AS c FROM kinds GROUP BY f)) AS row_groups_per_file
          ), x AS (
            SELECT *,
                   least(row_groups, 1 + COALESCE(rg_overlaps, 0) + rg_no_stats) AS rg_read,
                   least(files, 1 + COALESCE(file_overlaps, 0) + files_no_stats) AS files_read
            FROM m
          )
          SELECT *, 100.0 * rg_read / row_groups AS read_pct,
                 100.0 * files_read / files AS files_read_pct,
                 100.0 / row_groups AS best_pct
          FROM x;

          -- The files whose range overlaps the most others, first.
          CREATE TABLE widest AS
          WITH top AS (
            SELECT id AS file_name, CAST(lo AS VARCHAR) AS min_value, CAST(hi AS VARCHAR) AS max_value, n_overlap AS other_files_overlapping
            FROM ov WHERE grain = 'file'
            ORDER BY n_overlap DESC, id
            LIMIT 50
          )
          SELECT t.file_name, (SELECT count(*) FROM kinds k WHERE k.f = t.file_name) AS row_groups,
                 t.min_value, t.max_value, t.other_files_overlapping
          FROM top t
          ORDER BY t.other_files_overlapping DESC, t.file_name;

          COPY widest TO '{{ outputFiles.widest_files }}' (HEADER, DELIMITER ',');

          WITH d AS (
            SELECT *,
                   CASE WHEN best_pct > {{ inputs.max_read_pct }} THEN 'TOO_FEW_ROW_GROUPS'
                        WHEN read_pct > {{ inputs.max_read_pct }} THEN 'SKIPPING_POOR'
                        ELSE 'SKIPPING_OK' END AS outcome
            FROM r
          )
          SELECT outcome,
                 'Column {{ inputs.column }}: a lookup may read up to about ' || round(rg_read, 1) || ' of ' || row_groups || ' row groups ('
                   || round(read_pct, 1) || ' percent) and ' || round(files_read, 1) || ' of ' || files || ' files ('
                   || round(files_read_pct, 1) || ' percent). '
                   || CASE outcome
                        WHEN 'SKIPPING_OK' THEN 'That is within the target of ' || {{ inputs.max_read_pct }} || ' percent.'
                        WHEN 'SKIPPING_POOR' THEN 'The target is at most ' || {{ inputs.max_read_pct }} || ' percent, and a perfect sort would give about ' || round(best_pct, 1) || ' percent.'
                        ELSE 'Even a perfect layout would read ' || round(best_pct, 1) || ' percent, above the target of ' || {{ inputs.max_read_pct }} || ' percent, so the layout is not judged. Audit more files or raise max_read_pct.'
                      END
                   || CASE WHEN rg_no_stats > 0 THEN ' ' || rg_no_stats || ' row group(s) have no statistics and count as always read.' ELSE '' END AS reason,
                 '{{ inputs.column }}' AS column_name,
                 '{{ (outputs.inspect_column.outputs | last).row.column_type }}' AS column_type,
                 files, row_groups, rg_no_stats, rg_no_values,
                 round(rg_read, 1) AS row_groups_read, round(read_pct, 1) AS read_pct,
                 round(files_read, 1) AS files_read, round(files_read_pct, 1) AS files_read_pct,
                 rg_overlaps_max AS most_overlaps, round(best_pct, 1) AS best_pct,
                 row_group_size, row_groups_per_file,
                 CASE WHEN outcome = 'SKIPPING_POOR' THEN
                      'Run SET threads = 1 first, because parallel writers interleave the files and the order is lost. Then: COPY (SELECT * FROM read_parquet('
                      || chr(39) || '{{ (outputs.find_files.outputs | last).row.lake_glob }}' || chr(39)
                      || ', hive_partitioning = false, union_by_name = true) ORDER BY "{{ inputs.column }}") TO ' || chr(39) || '<new folder>' || chr(39)
                      || ' (FORMAT parquet, ROW_GROUP_SIZE ' || row_group_size || ', ROW_GROUPS_PER_FILE ' || row_groups_per_file || ')'
                      ELSE '' END AS suggestion
          FROM d;

      - id: rewrite_copy
        type: io.kestra.plugin.jdbc.duckdb.Queries
        runIf: "{{ inputs.mode == 'REWRITE_COPY' and (outputs.audit.outputs |
          last).row.outcome == 'SKIPPING_POOR' }}"
        description: >
          Write the same rows sorted by the column to a new folder, with the row
          group size and the row groups per file of the original. The original
          files are only read. One thread, because parallel writers interleave
          the files and the sorted order is lost. verify_copy checks the result.
        sql: |
          SET threads = 1;

          COPY (
            SELECT * FROM read_parquet('{{ (outputs.find_files.outputs | last).row.lake_glob }}', hive_partitioning = false, union_by_name = true)
            ORDER BY "{{ inputs.column }}"
          ) TO '{{ (outputs.find_files.outputs | last).row.target_dir }}'
            (FORMAT parquet,
             ROW_GROUP_SIZE {{ (outputs.audit.outputs | last).row.row_group_size }},
             ROW_GROUPS_PER_FILE {{ (outputs.audit.outputs | last).row.row_groups_per_file }});

      - id: verify_copy
        type: io.kestra.plugin.jdbc.duckdb.Queries
        runIf: "{{ inputs.mode == 'REWRITE_COPY' and (outputs.audit.outputs |
          last).row.outcome == 'SKIPPING_POOR' }}"
        description: >
          Prove the copy before anyone uses it: same number of rows, same
          checksum of every row, then the same measurement as audit on the copy.
          A copy that differs in a single row is COPY_MISMATCH and fails the
          run.
        fetchType: FETCH_ONE
        sql: |
          -- Same rows: a count and the sum of a hash of every row, which does not depend on the order.
          CREATE TABLE src_sum AS
          SELECT count(*) AS n, sum(CAST(hash(t) AS HUGEINT)) AS h
          FROM read_parquet('{{ (outputs.find_files.outputs | last).row.lake_glob }}', hive_partitioning = false, union_by_name = true) t;

          CREATE TABLE copy_sum AS
          SELECT count(*) AS n, sum(CAST(hash(t) AS HUGEINT)) AS h,
                 (SELECT count(*) FROM parquet_file_metadata('{{ (outputs.find_files.outputs | last).row.target_dir }}/*.parquet') WHERE num_rows = 0) AS empty_files
          FROM read_parquet('{{ (outputs.find_files.outputs | last).row.target_dir }}/*.parquet', hive_partitioning = false, union_by_name = true) t;

          -- Footer scan of the copy.
          CREATE TABLE rg AS
          SELECT file_name AS f, row_group_id AS g, row_group_num_rows AS n, num_values, stats_null_count AS nulls,
                 TRY_CAST(stats_min_value AS {{ (outputs.inspect_column.outputs | last).row.column_type }}) AS lo,
                 TRY_CAST(stats_max_value AS {{ (outputs.inspect_column.outputs | last).row.column_type }}) AS hi
          FROM parquet_metadata('{{ (outputs.find_files.outputs | last).row.target_dir }}/*.parquet')
          WHERE lower(path_in_schema) = lower('{{ inputs.column }}');

          -- Measurement. The same as in audit.
          CREATE TABLE kinds AS
          SELECT f, g, n, lo, hi,
                 CASE WHEN COALESCE(nulls, -1) = num_values THEN 'NO_VALUES'
                      WHEN lo IS NULL OR hi IS NULL THEN 'NO_STATS'
                      ELSE 'RANGED' END AS kind
          FROM rg;

          CREATE TABLE units AS
          SELECT 'row_group' AS grain, f || '#' || g AS id, lo, hi FROM kinds WHERE kind = 'RANGED'
          UNION ALL
          SELECT 'file', f, min(lo), max(hi) FROM kinds
          WHERE kind = 'RANGED' AND f NOT IN (SELECT f FROM kinds WHERE kind = 'NO_STATS')
          GROUP BY f;

          CREATE TABLE ov AS
          WITH s AS (
            SELECT grain, 'A' AS pass, id, lo AS v, 0 AS ord, 1 AS step, 0 AS ask, lo AS u_lo, hi AS u_hi FROM units
            UNION ALL
            SELECT grain, 'A', id, hi, 1, 0, 1, lo, hi FROM units
            UNION ALL
            SELECT grain, 'B', id, lo, 0, 0, 1, lo, hi FROM units
            UNION ALL
            SELECT grain, 'B', id, hi, 1, 1, 0, lo, hi FROM units
          ), c AS (
            SELECT grain, pass, id, ask, u_lo, u_hi,
                   sum(step) OVER (PARTITION BY grain, pass ORDER BY v, ord ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running
            FROM s
          )
          SELECT grain, id, min(u_lo) AS lo, min(u_hi) AS hi,
                 CAST(max(CASE WHEN pass = 'A' THEN running END) - max(CASE WHEN pass = 'B' THEN running END) - 1 AS BIGINT) AS n_overlap
          FROM c
          WHERE ask = 1
          GROUP BY grain, id;

          CREATE TABLE r AS
          WITH m AS (
            SELECT (SELECT count(*) FROM kinds WHERE kind <> 'NO_VALUES') AS row_groups,
                   (SELECT count(DISTINCT f) FROM kinds WHERE kind <> 'NO_VALUES') AS files,
                   (SELECT count(*) FROM kinds WHERE kind = 'NO_STATS') AS rg_no_stats,
                   (SELECT count(DISTINCT f) FROM kinds WHERE kind = 'NO_STATS') AS files_no_stats,
                   (SELECT avg(n_overlap) FROM ov WHERE grain = 'row_group') AS rg_overlaps,
                   (SELECT avg(n_overlap) FROM ov WHERE grain = 'file') AS file_overlaps
          ), x AS (
            SELECT *,
                   least(row_groups, 1 + COALESCE(rg_overlaps, 0) + rg_no_stats) AS rg_read,
                   least(files, 1 + COALESCE(file_overlaps, 0) + files_no_stats) AS files_read
            FROM m
          )
          SELECT *, 100.0 * rg_read / row_groups AS read_pct,
                 100.0 * files_read / files AS files_read_pct
          FROM x;

          WITH d AS (
            SELECT r.*, s.n AS rows_source, c.n AS rows_copy, c.empty_files, (s.n = c.n AND s.h = c.h) AS identical
            FROM r, src_sum s, copy_sum c
          ), o AS (
            SELECT *,
                   CASE WHEN NOT identical THEN 'COPY_MISMATCH'
                        WHEN read_pct <= {{ inputs.max_read_pct }} THEN 'FIXED'
                        ELSE 'COPY_BELOW_TARGET' END AS outcome
            FROM d
          )
          SELECT outcome,
                 CASE outcome
                   WHEN 'COPY_MISMATCH' THEN 'The copy in {{ (outputs.find_files.outputs | last).row.target_dir }} does not match the source: ' || rows_source || ' rows against ' || rows_copy || ', or a different checksum. Do not use it.'
                   ELSE 'The sorted copy in {{ (outputs.find_files.outputs | last).row.target_dir }} has the same ' || rows_source || ' rows and the same checksum. A lookup on {{ inputs.column }} may now read up to about '
                        || round(rg_read, 1) || ' of ' || row_groups || ' row groups (' || round(read_pct, 1) || ' percent) and ' || round(files_read, 1) || ' of ' || files
                        || ' files, down from {{ (outputs.audit.outputs | last).row.read_pct }} percent of the row groups. '
                        || CASE outcome WHEN 'FIXED' THEN 'That is within the target of ' || {{ inputs.max_read_pct }} || ' percent.'
                                        ELSE 'The target is ' || {{ inputs.max_read_pct }} || ' percent, so the copy is better but does not reach it. A column with few distinct values, or a copy with few row groups (' || row_groups || ' here), cannot get lower.' END
                        || ' Your files were not changed.'
                        || CASE WHEN empty_files > 0 THEN ' DuckDB left ' || empty_files || ' empty file(s) at the end of the copy. They are harmless and can be deleted.' ELSE '' END
                 END AS reason,
                 rows_source, rows_copy, identical, empty_files,
                 CAST({{ (outputs.audit.outputs | last).row.read_pct }} AS DOUBLE) AS read_pct_before,
                 round(read_pct, 1) AS read_pct_after,
                 round(files_read_pct, 1) AS files_read_pct_after,
                 row_groups AS row_groups_copy, files AS files_copy
          FROM o;

  - id: result
    type: io.kestra.plugin.core.output.OutputValues
    description: The row of verify_copy when a sorted copy was made, otherwise the
      row of audit.
    values:
      row: "{{ ((inputs.mode == 'REWRITE_COPY' and (outputs.audit.outputs |
        last).row.outcome == 'SKIPPING_POOR') ? (outputs.verify_copy.outputs |
        last).row : (outputs.audit.outputs | last).row) | toJson }}"

  - id: log_result
    type: io.kestra.plugin.core.log.Log
    message: "{{ outputs.result.values.row | jq('.reason') | first }}"

  - id: route
    type: io.kestra.plugin.core.flow.Switch
    description: >
      SKIPPING_OK, TOO_FEW_ROW_GROUPS and FIXED log and end. SKIPPING_POOR and
      COPY_BELOW_TARGET warn, and fail the run when on_poor_skipping is FAIL.
      COPY_MISMATCH always fails, because the copy must not be used.
    value: "{{ outputs.result.values.row | jq('.outcome') | first }}"
    cases:
      SKIPPING_OK:
        - id: log_ok
          type: io.kestra.plugin.core.log.Log
          message: "No action needed for column {{ inputs.column }}."

      TOO_FEW_ROW_GROUPS:
        - id: log_too_few
          type: io.kestra.plugin.core.log.Log
          level: WARN
          message: "Too few row groups to judge column {{ inputs.column }}. Nothing to fix
            yet."

      SKIPPING_POOR:
        - id: log_poor
          type: io.kestra.plugin.core.log.Log
          level: WARN
          message: >-
            The layout does not skip well for column {{ inputs.column }}. To fix
            it: {{ outputs.result.values.row | jq('.suggestion') | first }}. The
            files that overlap the most are in
            outputs.audit.outputFiles.widest_files.

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

      FIXED:
        - id: log_fixed
          type: io.kestra.plugin.core.log.Log
          message: "The sorted copy is verified and reaches the target. Compare it, then
            swap it in with your own tooling."

      COPY_BELOW_TARGET:
        - id: log_below
          type: io.kestra.plugin.core.log.Log
          level: WARN
          message: "The sorted copy is verified and better, but it does not reach
            max_read_pct. A column with few distinct values cannot get lower."

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

      COPY_MISMATCH:
        - id: fail_mismatch
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: "COPY_MISMATCH: {{ outputs.result.values.row | jq('.reason') |
            first }}"

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

errors:
  - id: report_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: "Data skipping audit failed in {{ flow.id }}, execution {{ execution.id
      }}. A BLOCKED_ message means the files, the column or the copy folder
      cannot be used as given and nothing was audited. SKIPPING_POOR or
      COPY_BELOW_TARGET means on_poor_skipping is FAIL. COPY_MISMATCH means the
      sorted copy differs from the source and must not be used. Otherwise read
      the task logs: an unreadable Parquet file stops the scan and its path is
      in the error."