Extract a CSV from HTTP and bulk load it into Oracle

id: oracle-extract-load-batch
namespace: company.team

inputs:
  - id: source_url
    type: STRING
    defaults: https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv
  - id: target_table
    type: SELECT
    description: Destination staging table. A SELECT input keeps the table name on
      an allow-list so a free-text value can never be concatenated into the DDL
      or INSERT.
    defaults: ORDERS_RAW
    values:
      - ORDERS_RAW
      - ORDERS_RAW_BACKFILL

variables:
  oracle_url: jdbc:oracle:thin:@//oracle.internal:1521/ORCLPDB1

concurrency:
  limit: 1

labels:
  team: data
  pipeline: orders-el

pluginDefaults:
  - type: io.kestra.plugin.jdbc.oracle
    values:
      url: "{{ vars.oracle_url }}"
      username: "{{ secret('ORACLE_USERNAME') }}"
      password: "{{ secret('ORACLE_PASSWORD') }}"

tasks:
  - id: download_csv
    type: io.kestra.plugin.core.http.Download
    description: Extract the source CSV from the upstream HTTP endpoint into
      internal storage.
    uri: "{{ inputs.source_url }}"
    saveAs: orders.csv
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3

  - id: csv_to_ion
    type: io.kestra.plugin.serdes.csv.CsvToIon
    description: Convert the downloaded CSV into the ION format that the Oracle
      Batch task consumes.
    from: "{{ outputs.download_csv.uri }}"

  - id: create_table
    type: io.kestra.plugin.jdbc.oracle.Query
    description: Create the staging table if it does not already exist before loading rows.
    fetchType: NONE
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3
    sql: |
      BEGIN
        EXECUTE IMMEDIATE 'CREATE TABLE {{ inputs.target_table }} (
          order_id      NUMBER,
          customer_name VARCHAR2(200),
          product       VARCHAR2(200),
          price         NUMBER(10,2),
          quantity      NUMBER,
          total         NUMBER(12,2)
        )';
      EXCEPTION
        WHEN OTHERS THEN
          IF SQLCODE != -955 THEN
            RAISE;
          END IF;
      END;

  - id: truncate_staging
    type: io.kestra.plugin.jdbc.oracle.Query
    description: Truncate the staging table so a re-run or backfill reloads cleanly
      instead of appending the CSV again.
    fetchType: NONE
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3
    sql: "TRUNCATE TABLE {{ inputs.target_table }}"

  - id: bulk_insert
    type: io.kestra.plugin.jdbc.oracle.Batch
    description: Bulk insert the ION rows into the freshly truncated staging table
      using named-column JDBC batches.
    from: "{{ outputs.csv_to_ion.uri }}"
    chunk: 1000
    columns:
      - order_id
      - customer_name
      - product
      - price
      - quantity
      - total
    sql: "INSERT INTO {{ inputs.target_table }} (order_id, customer_name, product,
      price, quantity, total) VALUES (?, ?, ?, ?, ?, ?)"

  - id: verify_load
    type: io.kestra.plugin.jdbc.oracle.Query
    description: Read back the row count so the assertion can compare it against the
      inserted count.
    fetchType: FETCH_ONE
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3
    sql: "SELECT COUNT(*) AS row_count FROM {{ inputs.target_table }}"

  - id: assert_load
    type: io.kestra.plugin.core.execution.Assert
    description: Fail loudly when the load landed zero rows or the table count does
      not match the rows just inserted.
    errorMessage: "Load verification failed for {{ inputs.target_table }}: inserted
      {{ outputs.bulk_insert.updatedCount }}, table reports {{
      outputs.verify_load.row.ROW_COUNT }}."
    conditions:
      - "{{ outputs.bulk_insert.updatedCount > 0 }}"
      - "{{ outputs.verify_load.row.ROW_COUNT ==
        outputs.bulk_insert.updatedCount }}"

  - id: log_summary
    type: io.kestra.plugin.core.log.Log
    description: Surface read vs inserted counts so silently dropped rows are
      obvious in the logs.
    message: "Read {{ outputs.bulk_insert.rowCount }} row(s) from the CSV and
      inserted {{ outputs.bulk_insert.updatedCount }} into {{
      inputs.target_table }}. Table now reports {{
      outputs.verify_load.row.ROW_COUNT }}."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack if the extract-load run fails so a broken nightly load
      is never silent.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    messageText: ":rotating_light: {{ flow.namespace }}.{{ flow.id }} failed
      (execution {{ execution.id }})."