wenshu/scripts/load_iceberg_demo.sh

#!/usr/bin/env bash
set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
ROOT_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"

read_dotenv_value() {
  local key="$1"
  local dotenv_file="${ROOT_DIR}/.env"
  if [[ ! -f "$dotenv_file" ]]; then
    return 0
  fi
  local raw
  raw="$(grep -m1 -E "^${key}=" "$dotenv_file" || true)"
  if [[ -z "$raw" ]]; then
    return 0
  fi
  printf '%s' "${raw#*=}"
}

DOTENV_BACKEND_URL="$(read_dotenv_value BACKEND_URL)"
BACKEND_URL="${BACKEND_URL:-${DOTENV_BACKEND_URL:-http://localhost:8080}}"
ICEBERG_DB="${ICEBERG_DB:-demo_iceberg}"
ICEBERG_USERNAME="${ICEBERG_USERNAME:-iceberg_demo}"
ICEBERG_PASSWORD="${ICEBERG_PASSWORD:-Demo1234}"
ICEBERG_EMAIL="${ICEBERG_EMAIL:-iceberg_demo@example.com}"
ICEBERG_ROLE="${ICEBERG_ROLE:-ANALYST}"
WORKFLOW_NAME="${WORKFLOW_NAME:-Iceberg_Sales_Clean_Analysis}"

json_field() {
  python3 - "$1" "$2" <<'PY'
import json
import sys

path = sys.argv[1]
field = sys.argv[2]
with open(path, "r", encoding="utf-8") as fh:
    data = json.load(fh)
value = data.get(field, "")
if value is None:
    value = ""
print(value)
PY
}

obtain_token() {
  local login_body login_code register_body register_code token
  login_body="$(mktemp)"
  login_code="$(curl -sS -o "$login_body" -w '%{http_code}' \
    -H 'Content-Type: application/json' \
    -d "{\"username\":\"${ICEBERG_USERNAME}\",\"password\":\"${ICEBERG_PASSWORD}\"}" \
    "${BACKEND_URL}/api/auth/login" || true)"

  if [[ "$login_code" != "200" ]]; then
    register_body="$(mktemp)"
    register_code="$(curl -sS -o "$register_body" -w '%{http_code}' \
      -H 'Content-Type: application/json' \
      -d "{\"username\":\"${ICEBERG_USERNAME}\",\"password\":\"${ICEBERG_PASSWORD}\",\"email\":\"${ICEBERG_EMAIL}\",\"role\":\"${ICEBERG_ROLE}\"}" \
      "${BACKEND_URL}/api/auth/register" || true)"
    if [[ "$register_code" != "200" ]]; then
      cat "$register_body" >&2
      rm -f "$login_body" "$register_body"
      echo "failed to register iceberg demo user" >&2
      exit 1
    fi
    rm -f "$register_body"

    login_body="$(mktemp)"
    login_code="$(curl -sS -o "$login_body" -w '%{http_code}' \
      -H 'Content-Type: application/json' \
      -d "{\"username\":\"${ICEBERG_USERNAME}\",\"password\":\"${ICEBERG_PASSWORD}\"}" \
      "${BACKEND_URL}/api/auth/login" || true)"
  fi

  if [[ "$login_code" != "200" ]]; then
    cat "$login_body" >&2
    rm -f "$login_body"
    echo "failed to login iceberg demo user" >&2
    exit 1
  fi

  token="$(json_field "$login_body" token)"
  rm -f "$login_body"
  if [[ -z "$token" ]]; then
    echo "login response did not contain a token" >&2
    exit 1
  fi
  printf '%s' "$token"
}

create_db() {
  local token="$1"
  local tmp code
  tmp="$(mktemp)"
  code="$(curl -sS -o "$tmp" -w '%{http_code}' \
    -H "Authorization: Bearer ${token}" \
    -H 'Content-Type: application/json' \
    -d "{\"database\":\"${ICEBERG_DB}\"}" \
    "${BACKEND_URL}/api/data/databases" || true)"
  if [[ "$code" != "200" && "$code" != "409" ]]; then
    cat "$tmp" >&2
    rm -f "$tmp"
    echo "failed to create iceberg namespace ${ICEBERG_DB}" >&2
    exit 1
  fi
  rm -f "$tmp"
}

import_csv() {
  local token="$1"
  local table="$2"
  local file_path="$3"
  curl -sS \
    -H "Authorization: Bearer ${token}" \
    -F "file=@${file_path}" \
    -F "mode=overwrite" \
    "${BACKEND_URL}/api/data/databases/${ICEBERG_DB}/tables/${table}/import/csv" >/dev/null
}

create_workflow_via_api() {
  local token="$1"
  local payload_file response_file code workflow_id

  payload_file="$(mktemp)"
  response_file="$(mktemp)"
  cat >"$payload_file" <<EOF
{
  "workflowName": "${WORKFLOW_NAME}",
  "description": "Import three related Iceberg tables, clean dirty records, and generate sales KPIs",
  "timeoutSeconds": 9000,
  "failureStrategy": "STOP",
  "tasks": [
    {
      "id": 1,
      "name": "Clean Customers",
      "taskType": "SPARK_SQL",
      "clusterType": "SPARK",
      "script": "CREATE OR REPLACE TABLE polaris.${ICEBERG_DB}.customer_clean USING iceberg AS SELECT customer_id, trim(full_name) AS full_name, lower(trim(email)) AS email, initcap(trim(city)) AS city, upper(trim(region)) AS region, to_date(signup_date) AS signup_date, upper(trim(loyalty_tier)) AS loyalty_tier, CASE WHEN lower(trim(status)) IN ('active','1','y') THEN 'ACTIVE' ELSE 'INACTIVE' END AS status, CAST(age AS INT) AS age FROM polaris.${ICEBERG_DB}.demo_customers WHERE customer_id IS NOT NULL AND trim(email) <> '' AND email LIKE '%@%'",
      "executorId": 3001,
      "timeoutSeconds": 1200,
      "retryTimes": 1
    },
    {
      "id": 2,
      "name": "Clean Products",
      "taskType": "SPARK_SQL",
      "clusterType": "SPARK",
      "script": "CREATE OR REPLACE TABLE polaris.${ICEBERG_DB}.product_clean USING iceberg AS SELECT product_id, upper(trim(sku)) AS sku, trim(product_name) AS product_name, initcap(lower(trim(category))) AS category, CAST(price AS DECIMAL(12,2)) AS price, CAST(stock_qty AS INT) AS stock_qty, to_date(launch_date) AS launch_date, CASE WHEN lower(trim(active_flag)) IN ('y','1','true') THEN true ELSE false END AS is_active FROM polaris.${ICEBERG_DB}.demo_products WHERE product_id IS NOT NULL AND price > 0 AND stock_qty >= 0",
      "executorId": 3001,
      "timeoutSeconds": 1200,
      "retryTimes": 1
    },
    {
      "id": 3,
      "name": "Clean Orders",
      "taskType": "SPARK_SQL",
      "clusterType": "SPARK",
      "script": "CREATE OR REPLACE TABLE polaris.${ICEBERG_DB}.order_clean USING iceberg AS SELECT order_id, customer_id, product_id, to_date(order_date) AS order_date, CAST(quantity AS INT) AS quantity, CAST(unit_price AS DECIMAL(12,2)) AS unit_price, CAST(COALESCE(discount_rate, 0) AS DECIMAL(5,4)) AS discount_rate, upper(trim(status)) AS status, upper(trim(channel)) AS channel, upper(trim(region)) AS region, NULLIF(trim(note), '') AS note FROM polaris.${ICEBERG_DB}.demo_orders WHERE order_id IS NOT NULL AND customer_id IS NOT NULL AND product_id IS NOT NULL AND quantity > 0 AND unit_price > 0",
      "executorId": 3001,
      "timeoutSeconds": 1500,
      "retryTimes": 1
    },
    {
      "id": 4,
      "name": "Build Enriched Fact",
      "taskType": "SPARK_SQL",
      "clusterType": "SPARK",
      "script": "CREATE OR REPLACE TABLE polaris.${ICEBERG_DB}.order_enriched USING iceberg AS SELECT o.order_id, o.order_date, o.customer_id, c.full_name, c.city, c.region AS customer_region, p.category, p.sku, o.channel, o.status, o.quantity, o.unit_price, o.discount_rate, CAST(o.quantity * o.unit_price AS DECIMAL(14,2)) AS gross_amount, CAST(o.quantity * o.unit_price * (1 - COALESCE(o.discount_rate, 0)) AS DECIMAL(14,2)) AS net_amount FROM polaris.${ICEBERG_DB}.order_clean o JOIN polaris.${ICEBERG_DB}.customer_clean c ON o.customer_id = c.customer_id JOIN polaris.${ICEBERG_DB}.product_clean p ON o.product_id = p.product_id",
      "executorId": 3001,
      "timeoutSeconds": 1800,
      "retryTimes": 1
    },
    {
      "id": 5,
      "name": "Build Daily KPI",
      "taskType": "SPARK_SQL",
      "clusterType": "SPARK",
      "script": "CREATE OR REPLACE TABLE polaris.${ICEBERG_DB}.sales_kpi_daily USING iceberg AS SELECT order_date, customer_region, category, COUNT(*) AS order_count, COUNT(DISTINCT customer_id) AS customer_count, SUM(quantity) AS total_quantity, ROUND(SUM(net_amount), 2) AS net_revenue, ROUND(AVG(net_amount), 2) AS avg_net_order_value, ROUND(MAX(net_amount), 2) AS max_net_order_value FROM polaris.${ICEBERG_DB}.order_enriched GROUP BY order_date, customer_region, category",
      "executorId": 3001,
      "timeoutSeconds": 1200,
      "retryTimes": 1
    }
  ],
  "edges": [
    { "from": 1, "to": 4 },
    { "from": 2, "to": 4 },
    { "from": 3, "to": 4 },
    { "from": 4, "to": 5 }
  ]
}
EOF

  code="$(curl -sS -o "$response_file" -w '%{http_code}' \
    -H "Authorization: Bearer ${token}" \
    -H 'Content-Type: application/json' \
    --data-binary "@${payload_file}" \
    "${BACKEND_URL}/api/task-build/code/create" || true)"

  rm -f "$payload_file"
  if [[ "$code" != "200" ]]; then
    cat "$response_file" >&2
    rm -f "$response_file"
    echo "failed to create workflow via /api/task-build/code/create" >&2
    exit 1
  fi

  workflow_id="$(json_field "$response_file" workflowId)"
  rm -f "$response_file"
  if [[ -z "$workflow_id" ]]; then
    echo "workflow create response missing workflowId" >&2
    exit 1
  fi
  echo "$workflow_id"
}

TOKEN="$(obtain_token)"
create_db "$TOKEN"
import_csv "$TOKEN" "demo_customers" "${ROOT_DIR}/backend/src/main/resources/db/iceberg_demo/demo_customers.csv"
import_csv "$TOKEN" "demo_products" "${ROOT_DIR}/backend/src/main/resources/db/iceberg_demo/demo_products.csv"
import_csv "$TOKEN" "demo_orders" "${ROOT_DIR}/backend/src/main/resources/db/iceberg_demo/demo_orders.csv"
WORKFLOW_ID="$(create_workflow_via_api "$TOKEN")"

echo "Imported demo tables into Iceberg namespace ${ICEBERG_DB}."
echo "Created demo workflow via /api/task-build/code/create, workflowId=${WORKFLOW_ID}."