Make GCS the trigger, not just the bucket.

id: bigquery-incremental-partitions
namespace: company.team
description: |
  Refresh only the BigQuery partitions that changed instead of rescanning the
  whole table. For each day in a small rolling window, the flow loads that
  day's GCS files into the matching partition with a partition decorator and
  WRITE_TRUNCATE, so each partition is atomically replaced and cost tracks the
  delta. Reruns are safe because each load truncates just its own partition.

inputs:
  - id: day_offsets
    type: JSON
    defaults: [ 0, -1, -2 ]
    description: Days to refresh, counting back from today (0 = today, -1 =
      yesterday). Covers your late-arriving-data window.

tasks:
  - id: refresh_window
    type: io.kestra.plugin.core.flow.ForEach
    description: Replace each day's partition from its own GCS path, one at a time.
    values: "{{ inputs.day_offsets }}"
    concurrencyLimit: 1
    tasks:
      - id: load_partition
        type: io.kestra.plugin.gcp.bigquery.LoadFromGcs
        description: Load one day's files into that day's partition, truncating just
          that partition.
        allowFailure: true
        from:
          - "gs://{{ secret('GCS_BUCKET') }}/events/dt={{ now() |
            dateAdd(taskrun.value, 'DAYS') | date('yyyy-MM-dd') }}/*.parquet"
        destinationTable: "{{ secret('GCP_PROJECT_ID') }}.analytics.events${{ now() |
          dateAdd(taskrun.value, 'DAYS') | date('yyyyMMdd') }}"
        format: PARQUET
        writeDisposition: WRITE_TRUNCATE

  - id: notify
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Confirm the incremental partition refresh.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "BigQuery partition refresh complete for {{ inputs.day_offsets | length }} day(s) (execution {{ execution.id }})."
      }

pluginDefaults:
  - type: io.kestra.plugin.gcp.bigquery
    values:
      projectId: "{{ secret('GCP_PROJECT_ID') }}"
      serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT') }}"

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Refresh the recent partitions on a cadence. Adjust or disable as needed.
    cron: "0 5 * * *"
    disabled: true

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack if the partition refresh fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "BigQuery partition refresh failed (execution {{ execution.id }})."
      }

id: gcs-download-transform-load
namespace: company.team
description: |
  Curate raw GCS files into a clean BigQuery table. On a schedule, the flow
  downloads a file from a GCS prefix, transforms it with Python and pandas in a
  container, loads the cleaned result into BigQuery, and notifies Discord. The
  classic download, transform, load pattern with object storage as the source.

inputs:
  - id: source_object
    type: STRING
    defaults: raw/customers.csv
    description: Object path inside the bucket to download and curate.

tasks:
  - id: download
    type: io.kestra.plugin.gcp.gcs.Download
    description: Pull the raw file from GCS into internal storage.
    from: "gs://{{ secret('GCS_BUCKET') }}/{{ inputs.source_object }}"

  - id: transform
    type: io.kestra.plugin.scripts.python.Script
    description: Clean and reshape the file with pandas, emit a curated CSV.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: ghcr.io/kestra-io/pydata:latest
    inputFiles:
      data.csv: "{{ outputs.download.uri }}"
    outputFiles:
      - curated.csv
    script: |
      import pandas as pd

      df = pd.read_csv("data.csv")
      df.columns = [c.strip().lower().replace(" ", "_") for c in df.columns]
      df = df.dropna(how="all").drop_duplicates()
      df.to_csv("curated.csv", index=False)

  - id: load_curated
    type: io.kestra.plugin.gcp.bigquery.Load
    description: Load the curated file into the clean BigQuery table.
    from: "{{ outputs.transform.outputFiles['curated.csv'] }}"
    destinationTable: "{{ secret('GCP_PROJECT_ID') }}.analytics.customers_clean"
    format: CSV
    autodetect: true
    csvOptions:
      skipLeadingRows: 1
    createDisposition: CREATE_IF_NEEDED
    writeDisposition: WRITE_TRUNCATE

  - id: notify
    type: io.kestra.plugin.discord.DiscordIncomingWebhook
    description: Confirm the curated load on Discord.
    url: "{{ secret('DISCORD_WEBHOOK_URL') }}"
    payload: |
      {
        "content": "Curated {{ inputs.source_object }} into BigQuery (execution {{ execution.id }})."
      }

pluginDefaults:
  - type: io.kestra.plugin.gcp
    values:
      projectId: "{{ secret('GCP_PROJECT_ID') }}"
      serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT') }}"

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Curate the source file on a cadence. Adjust or disable as needed.
    cron: "0 6 * * *"
    disabled: true

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.discord.DiscordIncomingWebhook
    description: Alert Discord if the curation fails.
    url: "{{ secret('DISCORD_WEBHOOK_URL') }}"
    payload: |
      {
        "content": "GCS curation pipeline failed (execution {{ execution.id }})."
      }