Incremental, idempotent lake ingestion from MinIO with DuckDB

id: minio-incremental-lake-ingest-duckdb
namespace: company.team

inputs:
  - id: bucket
    type: STRING
    displayName: Lake bucket
    defaults: data-lake

  - id: incoming_prefix
    type: STRING
    displayName: Landing prefix
    description: Producers drop Parquet files here. Only files that have not been
      ingested yet live here.
    defaults: raw/incoming/

  - id: processed_prefix
    type: STRING
    displayName: Processed prefix
    description: Ingested files are moved here, keeping their sub-path.
    defaults: raw/processed/

  - id: curated_prefix
    type: STRING
    displayName: Curated prefix
    description: Holds `events/` (row level, partitioned by event_day) and
      `daily_metrics/` (one file per event_day).
    defaults: lake/curated/

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post the result to Slack. Needs the SLACK_WEBHOOK_URL secret.
    defaults: false

variables:
  s3_secret: >-
    CREATE OR REPLACE SECRET lake (TYPE s3, KEY_ID '{{
    secret('MINIO_ACCESS_KEY_ID') }}', SECRET '{{ secret('MINIO_SECRET_KEY_ID')
    }}', ENDPOINT '{{ secret('MINIO_ENDPOINT') | split('//') | last }}',
    URL_STYLE 'path', USE_SSL {{ (secret('MINIO_ENDPOINT') | split('https:') |
    length) > 1 }}, REGION 'us-east-1');
  lake: "s3://{{ inputs.bucket }}/{{ inputs.curated_prefix }}"

concurrency:
  limit: 1

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Drain the landing zone every hour. Each run handles only what
      arrived since the last one.
    cron: "0 * * * *"

tasks:
  - id: list_incoming
    type: io.kestra.plugin.minio.List
    description: Freeze the batch. Every later step uses exactly these keys, so a
      file that lands mid-run waits for the next run.
    endpoint: "{{ secret('MINIO_ENDPOINT') }}"
    accessKeyId: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('MINIO_SECRET_KEY_ID') }}"
    bucket: "{{ inputs.bucket }}"
    prefix: "{{ inputs.incoming_prefix }}"
    regexp: ".*\\.parquet$"
    filter: FILES

  - id: has_batch
    type: io.kestra.plugin.core.flow.If
    description: Nothing new means nothing to do.
    condition: "{{ outputs.list_incoming.objects | length > 0 }}"
    then:
      - id: batch
        type: io.kestra.plugin.core.output.OutputValues
        description: The batch as a DuckDB list of S3 URLs.
        values:
          files: "[{% for o in outputs.list_incoming.objects %}'s3://{{ inputs.bucket
            }}/{{ o.key }}'{% if not loop.last %}, {% endif %}{% endfor %}]"

      - id: list_curated
        type: io.kestra.plugin.minio.List
        description: Check whether curated events already exist. DuckDB fails on a glob
          that matches no file, so the first run skips the duplicate check.
        endpoint: "{{ secret('MINIO_ENDPOINT') }}"
        accessKeyId: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('MINIO_SECRET_KEY_ID') }}"
        bucket: "{{ inputs.bucket }}"
        prefix: "{{ inputs.curated_prefix }}events/"
        regexp: ".*\\.parquet$"
        filter: FILES

      - id: profile
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: >
          Check the batch against the contract straight from S3: the exact
          column names, and every row parsing as the right type. Nothing is
          written.
        fetchType: FETCH_ONE
        sql: |
          INSTALL httpfs; LOAD httpfs;
          {{ render(vars.s3_secret) }}
          CREATE TEMP TABLE b AS SELECT * FROM read_parquet({{ outputs.batch.values.files }}, union_by_name = true);
          SELECT
            (SELECT list(column_name ORDER BY column_name) FROM (DESCRIBE b))
              = ['amount', 'country', 'event_id', 'event_time', 'user_id'] AS names_ok,
            (SELECT string_agg(column_name || ' ' || column_type, ', ' ORDER BY column_name) FROM (DESCRIBE b)) AS columns,
            (SELECT count(*) FROM b) AS rows,
            (SELECT count(*) FROM b WHERE try_cast(event_time AS TIMESTAMP) IS NULL OR try_cast(amount AS DOUBLE) IS NULL OR event_id IS NULL) AS unparseable;

      - id: contract
        type: io.kestra.plugin.core.flow.If
        description: Load only when the names match and every row parses. Otherwise
          leave the files where they are.
        condition: "{{ (outputs.profile.outputs | last).row.names_ok and
          (outputs.profile.outputs | last).row.unparseable == 0 }}"
        then:
          - id: load
            type: io.kestra.plugin.jdbc.duckdb.Queries
            description: >
              Append the new rows to curated events, skipping event_ids already
              ingested, then rebuild daily_metrics for every day the batch
              touched from all curated events, so late data and resent files
              never double count.
            fetchType: FETCH_ONE
            sql: |
              INSTALL httpfs; LOAD httpfs;
              {{ render(vars.s3_secret) }}
              CREATE TEMP TABLE b AS
                SELECT event_id::VARCHAR AS event_id, user_id::VARCHAR AS user_id, country::VARCHAR AS country,
                       amount::DOUBLE AS amount, event_time::TIMESTAMP AS event_time, event_time::TIMESTAMP::DATE AS event_day
                FROM read_parquet({{ outputs.batch.values.files }}, union_by_name = true)
                QUALIFY row_number() OVER (PARTITION BY event_id ORDER BY event_time) = 1;
              CREATE TEMP TABLE seen AS
              {% if outputs.list_curated.objects | length > 0 %}
                SELECT DISTINCT event_id FROM read_parquet('{{ render(vars.lake) }}events/**/*.parquet', hive_partitioning = true) WHERE event_id IN (SELECT event_id FROM b);
              {% else %}
                SELECT NULL::VARCHAR AS event_id WHERE false;
              {% endif %}
              CREATE TEMP TABLE fresh AS SELECT * FROM b WHERE event_id NOT IN (SELECT event_id FROM seen);
              COPY (SELECT * FROM fresh) TO '{{ render(vars.lake) }}events'
                (FORMAT PARQUET, PARTITION_BY (event_day), OVERWRITE_OR_IGNORE, FILENAME_PATTERN 'batch-{{ execution.id }}-{i}');
              COPY (
                SELECT event_day, country, count(*) AS events, count(DISTINCT user_id) AS active_users, round(sum(amount), 2) AS revenue
                FROM read_parquet('{{ render(vars.lake) }}events/**/*.parquet', hive_partitioning = true)
                WHERE event_day IN (SELECT DISTINCT event_day FROM fresh)
                GROUP BY event_day, country
              ) TO '{{ render(vars.lake) }}daily_metrics'
                (FORMAT PARQUET, PARTITION_BY (event_day), OVERWRITE_OR_IGNORE, FILENAME_PATTERN 'metrics');
              SELECT
                (SELECT count(*) FROM fresh) AS loaded,
                (SELECT count(*) FROM seen) AS duplicates_skipped,
                (SELECT coalesce(string_agg(DISTINCT event_day::VARCHAR, ', ' ORDER BY event_day::VARCHAR), '') FROM fresh) AS days_rebuilt;

          - id: archive
            type: io.kestra.plugin.core.flow.Loop
            description: Move each file of the frozen batch to the processed prefix, keeping
              its sub-path so equal file names never collide.
            values: "{{ outputs.list_incoming.objects | jq('[.[].key]') | first }}"
            concurrencyLimit: 4
            tasks:
              - id: archive_object
                type: io.kestra.plugin.minio.Copy
                endpoint: "{{ secret('MINIO_ENDPOINT') }}"
                accessKeyId: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
                secretKeyId: "{{ secret('MINIO_SECRET_KEY_ID') }}"
                delete: true
                from:
                  bucket: "{{ inputs.bucket }}"
                  key: "{{ item.value }}"
                to:
                  bucket: "{{ inputs.bucket }}"
                  key: "{{ inputs.processed_prefix }}{{ item.value | slice(inputs.incoming_prefix
                    | length, item.value | length) }}"

          - id: log_loaded
            type: io.kestra.plugin.core.log.Log
            message: >-
              Ingested {{ outputs.list_incoming.objects | length }} file(s): {{
              (outputs.load.outputs | last).row.loaded }} new rows, {{
              (outputs.load.outputs | last).row.duplicates_skipped }}
              already-ingested rows skipped, daily_metrics rebuilt for [{{
              (outputs.load.outputs | last).row.days_rebuilt }}].

          - id: notify_loaded
            type: io.kestra.plugin.core.flow.If
            condition: "{{ inputs.notify_slack }}"
            then:
              - id: slack_loaded
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {
                    "text": "Lake ingest: {{ (outputs.load.outputs | last).row.loaded }} new rows from {{ outputs.list_incoming.objects | length }} file(s), {{ (outputs.load.outputs | last).row.duplicates_skipped }} duplicates skipped, metrics rebuilt for {{ (outputs.load.outputs | last).row.days_rebuilt }}. Execution {{ execution.id }}."
                  }
        else:
          - id: notify_contract
            type: io.kestra.plugin.core.flow.If
            condition: "{{ inputs.notify_slack }}"
            then:
              - id: slack_contract
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {
                    "text": "Lake ingest BLOCKED by the contract. Columns: {{ (outputs.profile.outputs | last).row.columns }}. Unparseable rows: {{ (outputs.profile.outputs | last).row.unparseable }}. Files left in {{ inputs.incoming_prefix }}. Execution {{ execution.id }}."
                  }

          - id: block_batch
            type: io.kestra.plugin.core.execution.Fail
            description: End the run as FAILED. The batch stays in the landing zone for a
              fix and a rerun.
            errorMessage: "Contract failed. Columns: {{ (outputs.profile.outputs |
              last).row.columns }}. Unparseable rows: {{
              (outputs.profile.outputs | last).row.unparseable }}."
    else:
      - id: nothing_new
        type: io.kestra.plugin.core.log.Log
        message: "No new files under s3://{{ inputs.bucket }}/{{ inputs.incoming_prefix
          }}."

errors:
  - id: notify_error
    type: io.kestra.plugin.core.flow.If
    description: A task failed outside the contract check, for example the object
      store was unreachable. The contract failure already posts its own message.
    condition: "{{ inputs.notify_slack and (outputs.block_batch ?? null) is null }}"
    then:
      - id: slack_error
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Lake ingest ERRORED in {{ flow.id }}. Files not yet archived stay in {{ inputs.incoming_prefix }} and are retried next run. Execution {{ execution.id }}."
          }