Add Gravitino metadata integration

Author: CHEUKFUNGWU

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test lint pipeline infra-light infra-serving infra-dashboard infra-stop init-superset dashboard-stop seed-data flink-up flink-submit flink-jobs flink-savepoint flink-cancel flink-restore batch-backfill sync-doris health demo demo-streaming demo-serving clean
+.PHONY: test lint pipeline infra-light infra-serving infra-dashboard infra-stop init-superset dashboard-stop seed-data flink-up flink-submit flink-jobs flink-savepoint flink-cancel flink-restore batch-backfill sync-doris health demo demo-streaming demo-serving gravitino-up gravitino-status gravitino-catalogs clean
 
 test:
 	uv run pytest -v
@@ -10,9 +10,10 @@ pipeline:
 	uv run python -m scripts.spark_paimon_backfill --input data/raw/mock_llm_requests/events.jsonl
 
 infra-light:
-	docker compose up -d postgres kafka flink-jobmanager flink-taskmanager
+	docker compose up -d postgres kafka gravitino flink-jobmanager flink-taskmanager
 	scripts/prepare_flink_warehouse.sh
 	scripts/create_kafka_topics.sh
+	scripts/init_gravitino.sh
 
 infra-serving:
 	docker compose up -d doris-fe doris-be doris-init
@@ -90,5 +91,15 @@ demo-streaming:
 demo-serving:
 	scripts/run_serving_demo.sh
 
+gravitino-up:
+	docker compose up -d --wait gravitino
+	scripts/init_gravitino.sh
+
+gravitino-status:
+	curl -fsS -H 'Accept: application/vnd.gravitino.v1+json' http://localhost:8090/api/version | python3 -m json.tool
+
+gravitino-catalogs:
+	curl -fsS -H 'Accept: application/vnd.gravitino.v1+json' http://localhost:8090/api/metalakes/ai_observability/catalogs | python3 -m json.tool
+
 clean:
 	rm -rf data/

README.md

@@ -210,9 +210,10 @@ Build and start the local Flink runtime manually:
 
 ```bash
 docker compose build flink-jobmanager
-docker compose up -d postgres kafka flink-jobmanager flink-taskmanager
+docker compose up -d postgres kafka gravitino flink-jobmanager flink-taskmanager
 scripts/prepare_flink_warehouse.sh
 scripts/create_kafka_topics.sh
+scripts/init_gravitino.sh
 ```
 
 Load local JSONL events into the Postgres operational source table:
@@ -248,6 +249,26 @@ The Flink Web UI is available at:
 http://localhost:8081
 ```
 
+Gravitino manages the shared Paimon metadata under the
+`ai_observability` metalake. Start and initialize it independently with:
+
+```bash
+make gravitino-up
+```
+
+Verify the server and registered catalogs with:
+
+```bash
+make gravitino-status
+make gravitino-catalogs
+```
+
+The Gravitino Web V2 UI is available at:
+
+```text
+http://localhost:8090
+```
+
 ## 9. Spark Backfill and Validation
 
 Spark is no longer a parallel warehouse pipeline. Its role is to backfill and validate the shared Paimon warehouse that Flink also writes to.

docker-compose.yml

@@ -20,6 +20,30 @@ services:
       - postgres_data:/var/lib/postgresql/data
       - ./sql/source_postgres_schema.sql:/docker-entrypoint-initdb.d/01_source_postgres_schema.sql:ro
 
+  gravitino:
+    image: apache/gravitino:1.2.0
+    container_name: ai-observability-gravitino
+    restart: unless-stopped
+    ports:
+      - "8090:8090"
+    environment:
+      GRAVITINO_MEM: "-Xms256m -Xmx512m -XX:MaxMetaspaceSize=256m"
+      GRAVITINO_USE_WEB_V2: "true"
+    volumes:
+      - gravitino_data:/root/gravitino/data
+      - paimon_warehouse:/workspace/data/paimon
+    healthcheck:
+      test:
+        - CMD-SHELL
+        - >-
+          curl -fsS
+          -H 'Accept: application/vnd.gravitino.v1+json'
+          http://localhost:8090/api/version >/dev/null
+      interval: 5s
+      timeout: 3s
+      retries: 24
+      start_period: 20s
+
   flink-jobmanager:
     build:
       context: .
@@ -210,6 +234,7 @@ services:
 
 volumes:
   postgres_data:
+  gravitino_data:
   paimon_warehouse:
   doris_fe_data:
   doris_be_data:

docs/runtime_runbook.md

@@ -15,6 +15,7 @@ These services are infrastructure processes. They keep running until stopped:
 | --- | --- | --- |
 | `postgres` | Operational source database with logical WAL enabled | `make infra-light` |
 | `kafka` | Real-time ODS buffer and replay log | `make infra-light` |
+| `gravitino` | Metadata service and Web UI for the shared Paimon catalog | `make infra-light` or `make gravitino-up` |
 | `flink-jobmanager` | Flink control plane and REST API | `make infra-light` |
 | `flink-taskmanager` | Flink worker slots for streaming SQL jobs | `make infra-light` |
 | `doris-fe` | Doris SQL frontend, only needed for serving checks | `make infra-serving` |
@@ -73,6 +74,25 @@ Avoid repeatedly running `make flink-submit` against an already-running cluster
 unless you intentionally want duplicate streaming jobs. Check `make flink-jobs`
 first.
 
+### Gravitino metadata service
+
+Start only Gravitino and initialize the `ai_observability` metalake plus the
+`paimon_lake` catalog:
+
+```bash
+make gravitino-up
+```
+
+The initializer is idempotent, so it is safe to rerun after a container restart.
+Check the API and catalog registration with:
+
+```bash
+make gravitino-status
+make gravitino-catalogs
+```
+
+Open the Web V2 UI at `http://localhost:8090`.
+
 ### Finite batch and serving tasks
 
 These commands are finite tasks. They should start, complete, and exit:

flink/sql/30_build_dws_from_dwd.sql

@@ -194,7 +194,8 @@ WITH session_metrics AS (
         session_id,
         COUNT(*) AS turn_cnt,
         SUM(total_tokens) AS token_cnt,
-        TIMESTAMPDIFF(MILLISECOND, MIN(created_at), MAX(created_at)) + MAX(latency_ms) AS session_duration_ms
+        TIMESTAMPDIFF(SECOND, MIN(created_at), MAX(created_at)) * 1000
+            + MAX(latency_ms) AS session_duration_ms
     FROM paimon_lake.dwd.dwd_ai_llm_request_di
     GROUP BY `date`, app_name, feature_name, session_id
 ),

scripts/init_gravitino.sh

@@ -0,0 +1,99 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+GRAVITINO_URL="${GRAVITINO_URL:-http://localhost:8090}"
+METALAKE="${GRAVITINO_METALAKE:-ai_observability}"
+RESPONSE_FILE="$(mktemp)"
+trap 'rm -f "${RESPONSE_FILE}"' EXIT
+
+headers=(
+  -H "Accept: application/vnd.gravitino.v1+json"
+  -H "Content-Type: application/json"
+)
+curl_timeout=(--connect-timeout 3 --max-time 10)
+
+get_status() {
+  curl -sS "${curl_timeout[@]}" -o "${RESPONSE_FILE}" -w '%{http_code}' \
+    "${headers[@]}" "$1"
+}
+
+create_resource() {
+  local label="$1"
+  local url="$2"
+  local payload="$3"
+  local status
+
+  status="$(
+    curl -sS "${curl_timeout[@]}" -o "${RESPONSE_FILE}" -w '%{http_code}' \
+      -X POST "${headers[@]}" -d "${payload}" "${url}"
+  )"
+  if [[ "${status}" != "200" && "${status}" != "201" ]]; then
+    echo "ERROR: failed to create ${label} (HTTP ${status})" >&2
+    cat "${RESPONSE_FILE}" >&2
+    return 1
+  fi
+  echo "Created ${label}."
+}
+
+ready=false
+for attempt in $(seq 1 60); do
+  if curl -fsS "${curl_timeout[@]}" "${headers[@]}" \
+    "${GRAVITINO_URL}/api/version" > /dev/null 2>&1; then
+    ready=true
+    break
+  fi
+  echo "Waiting for Gravitino server... (${attempt}/60)"
+  sleep 2
+done
+
+if [[ "${ready}" != "true" ]]; then
+  echo "ERROR: Gravitino server did not become ready within 120 seconds" >&2
+  exit 1
+fi
+
+metalake_status="$(get_status "${GRAVITINO_URL}/api/metalakes/${METALAKE}")"
+case "${metalake_status}" in
+  200)
+    echo "Metalake ${METALAKE} already exists."
+    ;;
+  404)
+    create_resource "metalake" "${GRAVITINO_URL}/api/metalakes" "$(cat <<JSON
+{
+  "name": "${METALAKE}",
+  "comment": "AI observability lakehouse metadata"
+}
+JSON
+)"
+    ;;
+  *)
+    echo "ERROR: failed to inspect metalake ${METALAKE} (HTTP ${metalake_status})" >&2
+    cat "${RESPONSE_FILE}" >&2
+    exit 1
+    ;;
+esac
+
+catalog_status="$(get_status "${GRAVITINO_URL}/api/metalakes/${METALAKE}/catalogs/paimon_lake")"
+case "${catalog_status}" in
+  200)
+    echo "Paimon catalog paimon_lake already exists."
+    ;;
+  404)
+    create_resource "Paimon catalog" "${GRAVITINO_URL}/api/metalakes/${METALAKE}/catalogs" '{
+      "name": "paimon_lake",
+      "type": "RELATIONAL",
+      "provider": "lakehouse-paimon",
+      "comment": "Paimon ODS, DWD, DWS, DIM, and ADS metadata",
+      "properties": {
+        "catalog-backend": "filesystem",
+        "warehouse": "file:///workspace/data/paimon"
+      }
+    }'
+    ;;
+  *)
+    echo "ERROR: failed to inspect Paimon catalog (HTTP ${catalog_status})" >&2
+    cat "${RESPONSE_FILE}" >&2
+    exit 1
+    ;;
+esac
+
+echo "Gravitino initialization complete: ${GRAVITINO_URL}"

scripts/run_flink_sql_sequence.sh

@@ -7,7 +7,8 @@ if [[ $# -lt 1 ]]; then
 fi
 
 tmp_file="flink/sql/.generated_sequence.sql"
-trap 'rm -f "${tmp_file}"' EXIT
+output_file="$(mktemp)"
+trap 'rm -f "${tmp_file}" "${output_file}"' EXIT
 rm -f "${tmp_file}"
 
 for sql_file in "$@"; do
@@ -20,6 +21,20 @@ done
 
 scripts/prepare_flink_warehouse.sh
 
+set +e
 docker compose run -T --rm flink-sql-client \
   /opt/flink/bin/sql-client.sh \
-  -f "/workspace/${tmp_file}"
+  -f "/workspace/${tmp_file}" 2>&1 | tee "${output_file}"
+docker_status=${PIPESTATUS[0]}
+set -e
+
+if [[ ${docker_status} -ne 0 ]]; then
+  exit "${docker_status}"
+fi
+
+# Flink SQL Client can print a statement-level error and still exit with status 0.
+# Treat any such error as a failed sequence so Make/CI cannot report false success.
+if grep -q '\[ERROR\]' "${output_file}"; then
+  echo "ERROR: Flink SQL Client reported one or more failed statements." >&2
+  exit 1
+fi

tests/test_flink_sql_assets.py

@@ -347,6 +347,15 @@ def test_flink_sql_sequence_runner_keeps_catalog_in_one_session():
     assert "docker compose run -T --rm flink-sql-client" in script
     assert "/opt/flink/bin/sql-client.sh" in script
     assert "-f \"/workspace/${tmp_file}\"" in script
+    assert "PIPESTATUS[0]" in script
+    assert "\\[ERROR\\]" in script
+
+
+def test_dws_session_duration_uses_a_flink_supported_timestampdiff_unit():
+    sql = read_asset("flink/sql/30_build_dws_from_dwd.sql")
+
+    assert "TIMESTAMPDIFF(MILLISECOND" not in sql
+    assert "TIMESTAMPDIFF(SECOND, MIN(created_at), MAX(created_at)) * 1000" in sql
 
 
 def test_flink_savepoint_restore_helpers_are_available():

tests/test_gravitino_assets.py

@@ -0,0 +1,44 @@
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def read_asset(relative_path: str) -> str:
+    return (REPO_ROOT / relative_path).read_text(encoding="utf-8")
+
+
+def test_compose_defines_a_health_checked_gravitino_service():
+    compose = read_asset("docker-compose.yml")
+
+    assert "gravitino:" in compose
+    assert "apache/gravitino:1.2.0" in compose
+    assert '"8090:8090"' in compose
+    assert "GRAVITINO_USE_WEB_V2: \"true\"" in compose
+    assert "healthcheck:" in compose
+    assert "http://localhost:8090/api/version" in compose
+    assert "gravitino_data:/root/gravitino/data" in compose
+    assert "paimon_warehouse:/workspace/data/paimon" in compose
+
+
+def test_gravitino_initializer_is_strict_and_idempotent():
+    script = read_asset("scripts/init_gravitino.sh")
+
+    assert "set -euo pipefail" in script
+    assert 'METALAKE="${GRAVITINO_METALAKE:-ai_observability}"' in script
+    assert 'get_status "${GRAVITINO_URL}/api/metalakes/${METALAKE}"' in script
+    assert 'create_resource "metalake"' in script
+    assert 'get_status "${GRAVITINO_URL}/api/metalakes/${METALAKE}/catalogs/paimon_lake"' in script
+    assert 'create_resource "Paimon catalog"' in script
+    assert "--connect-timeout 3 --max-time 10" in script
+    assert '|| echo "Metalake already exists"' not in script
+
+
+def test_makefile_exposes_gravitino_lifecycle_targets():
+    makefile = read_asset("Makefile")
+
+    assert "gravitino-up:" in makefile
+    assert "docker compose up -d --wait gravitino" in makefile
+    assert "scripts/init_gravitino.sh" in makefile
+    assert "gravitino-status:" in makefile
+    assert "gravitino-catalogs:" in makefile