update flink statement warnings UI based on new response structure

src/flinkSql/flinkStatementResultsManager.test.ts

@@ -262,6 +262,7 @@ describe("FlinkStatementResultsViewModel and FlinkStatementResultsManager", () =
       areResultsViewable: ctx.statement.canRequestResults,
       possiblyViewable: ctx.statement.possiblyViewable,
       isForeground: ctx.statement.isForeground,
+      warnings: ctx.statement.warnings,
     });
   });
 

src/flinkSql/flinkStatementResultsManager.ts

@@ -19,6 +19,7 @@ import type { ViewMode } from "./flinkStatementResultColumns";
 import type { StatementResultsRow } from "./flinkStatementResults";
 import { parseResults } from "./flinkStatementResults";
 import { extractPageToken } from "./utils";
+import type { StatementWarning } from "./warningParser";
 
 const logger = new Logger("flink-statement-results");
 
@@ -85,6 +86,7 @@ export type PostFunction = {
     failed: boolean;
     areResultsViewable: boolean;
     isForeground: boolean;
+    warnings: StatementWarning[];
   }>;
   (type: "StopStatement", body: { timestamp?: number }): Promise<null>;
   (type: "ViewStatementSource", body: { timestamp?: number }): Promise<null>;
@@ -493,6 +495,7 @@ export class FlinkStatementResultsManager {
           areResultsViewable: this.statement.canRequestResults,
           possiblyViewable: this.statement.possiblyViewable,
           isForeground: this.statement.isForeground,
+          warnings: this.statement.warnings,
         };
       }
       case "StopStatement": {

src/flinkSql/warningParser.test.ts

@@ -0,0 +1,147 @@
+import assert from "assert";
+import {
+  extractWarnings,
+  parseLegacyWarnings,
+  stripWarningsFromDetail,
+  type StatementWarning,
+} from "./warningParser";
+
+describe("flinkSql/warningParser", () => {
+  describe("parseLegacyWarnings", () => {
+    it("should return empty array for undefined input", () => {
+      const result = parseLegacyWarnings(undefined);
+      assert.deepStrictEqual(result, []);
+    });
+
+    it("should return empty array for empty string", () => {
+      const result = parseLegacyWarnings("");
+      assert.deepStrictEqual(result, []);
+    });
+
+    it("should return empty array when no [Warning] markers present", () => {
+      const result = parseLegacyWarnings("Statement is running successfully.");
+      assert.deepStrictEqual(result, []);
+    });
+
+    it("should parse a single legacy warning", () => {
+      const result = parseLegacyWarnings(
+        "[Warning] The primary key does not match the upsert key.",
+      );
+
+      assert.strictEqual(result.length, 1);
+      assert.strictEqual(result[0].severity, "MODERATE");
+      assert.strictEqual(result[0].reason, "");
+      assert.strictEqual(result[0].message, "The primary key does not match the upsert key.");
+      // created_at is null for legacy warnings (no timestamp available)
+      assert.strictEqual(result[0].created_at, null);
+    });
+
+    it("should parse multiple legacy warnings", () => {
+      const detail =
+        "[Warning] First warning message. [Warning] Second warning message with more details.";
+      const result = parseLegacyWarnings(detail);
+
+      assert.strictEqual(result.length, 2);
+      assert.strictEqual(result[0].message, "First warning message.");
+      assert.strictEqual(result[1].message, "Second warning message with more details.");
+    });
+
+    it("should handle real-world legacy warning format", () => {
+      const detail =
+        "[Warning] The primary key does not match the upsert key derived from the query. " +
+        "If the primary key and upsert key don't match, the system needs to add a state-intensive " +
+        "operation for correction. [Warning] Your query includes one or more highly state-intensive " +
+        "operators but does not set a time-to-live (TTL) value.";
+
+      const result = parseLegacyWarnings(detail);
+
+      assert.strictEqual(result.length, 2);
+      assert.ok(result[0].message.includes("primary key does not match"));
+      assert.ok(result[1].message.includes("state-intensive operators"));
+    });
+
+    it("should handle case-insensitive [Warning] markers", () => {
+      const result = parseLegacyWarnings("[warning] lowercase warning message.");
+
+      assert.strictEqual(result.length, 1);
+      assert.strictEqual(result[0].message, "lowercase warning message.");
+    });
+
+    it("should handle brackets within warning messages", () => {
+      const detail =
+        "[Warning] Please revisit the query (upsert key: [customer_name]) or provide a primary key.";
+      const result = parseLegacyWarnings(detail);
+
+      assert.strictEqual(result.length, 1);
+      assert.ok(result[0].message.includes("[customer_name]"));
+    });
+  });
+
+  describe("extractWarnings", () => {
+    it("should return empty array when both inputs are undefined", () => {
+      const result = extractWarnings(undefined, undefined);
+      assert.deepStrictEqual(result, []);
+    });
+
+    it("should prefer structured warnings when available", () => {
+      const apiWarnings: StatementWarning[] = [
+        {
+          severity: "CRITICAL",
+          created_at: new Date("2025-11-14T16:01:00Z"),
+          reason: "UPSERT_PRIMARY_KEY_MISMATCH",
+          message: "API warning message",
+        },
+      ];
+      const detail = "[Warning] Legacy warning message.";
+
+      const result = extractWarnings(apiWarnings, detail);
+
+      assert.strictEqual(result.length, 1);
+      assert.strictEqual(result[0].message, "API warning message");
+      assert.strictEqual(result[0].severity, "CRITICAL");
+    });
+
+    it("should fall back to legacy parsing when structured warnings empty", () => {
+      const result = extractWarnings([], "[Warning] Legacy warning.");
+
+      assert.strictEqual(result.length, 1);
+      assert.strictEqual(result[0].message, "Legacy warning.");
+    });
+
+    it("should fall back to legacy parsing when structured warnings undefined", () => {
+      const result = extractWarnings(undefined, "[Warning] Legacy warning.");
+
+      assert.strictEqual(result.length, 1);
+      assert.strictEqual(result[0].message, "Legacy warning.");
+    });
+
+    it("should return empty array when detail has no warnings", () => {
+      const result = extractWarnings(undefined, "Statement running successfully.");
+      assert.deepStrictEqual(result, []);
+    });
+  });
+
+  describe("stripWarningsFromDetail", () => {
+    it("should return null for undefined input", () => {
+      const result = stripWarningsFromDetail(undefined);
+      assert.strictEqual(result, null);
+    });
+
+    it("should return null when detail contains only warnings", () => {
+      const result = stripWarningsFromDetail("[Warning] First. [Warning] Second.");
+      assert.strictEqual(result, null);
+    });
+
+    it("should preserve non-warning content before warnings", () => {
+      const detail = "Statement running. [Warning] Some warning message.";
+      const result = stripWarningsFromDetail(detail);
+      assert.strictEqual(result, "Statement running.");
+    });
+
+    it("should handle detail with no warnings", () => {
+      const detail = "Statement completed successfully.";
+      const result = stripWarningsFromDetail(detail);
+      assert.strictEqual(result, "Statement completed successfully.");
+    });
+  });
+});

src/flinkSql/warningParser.ts

@@ -0,0 +1,79 @@
+/**
+ * Warning parsing utilities for Flink statements.
+ * Handles both new structured API warnings and legacy [Warning] format in detail strings.
+ */
+
+/** Regex matching `[Warning] message` blocks in legacy detail strings. */
+const LEGACY_WARNING_PATTERN = /\[Warning\]\s*([\s\S]*?)(?=\s*\[Warning\]|$)/gi;
+
+/** API warning shape from the new structured warnings field */
+export interface StatementWarning {
+  readonly severity: string;
+  readonly created_at: Date | null;
+  readonly reason: string;
+  readonly message: string;
+}
+
+/**
+ * Parse legacy [Warning] format from detail string.
+ * Legacy format: "[Warning] message. [Warning] another message."
+ * @param detail The detail string that may contain legacy warnings
+ * @returns Array of parsed StatementWarning objects
+ */
+export function parseLegacyWarnings(detail: string | undefined): StatementWarning[] {
+  if (!detail) {
+    return [];
+  }
+
+  const warnings: StatementWarning[] = [];
+  LEGACY_WARNING_PATTERN.lastIndex = 0;
+  let match: RegExpExecArray | null;
+
+  while ((match = LEGACY_WARNING_PATTERN.exec(detail)) !== null) {
+    const message = match[1].trim();
+    if (message) {
+      warnings.push({
+        severity: "MODERATE",
+        created_at: null, // No timestamp in legacy format
+        reason: "", // No reason in legacy format
+        message,
+      });
+    }
+  }
+
+  return warnings;
+}
+
+/**
+ * Extract warnings from API response, preferring structured warnings over legacy parsing.
+ * @param warnings The structured warnings array from the API (may be undefined)
+ * @param detail The detail string that may contain legacy warnings
+ * @returns Array of parsed StatementWarning objects
+ */
+export function extractWarnings(
+  warnings: StatementWarning[] | undefined,
+  detail: string | undefined,
+): StatementWarning[] {
+  // Prefer structured warnings if available
+  if (warnings && warnings.length > 0) {
+    return warnings;
+  }
+
+  // Fall back to legacy parsing
+  return parseLegacyWarnings(detail);
+}
+
+/**
+ * Strip [Warning] sections from a detail string.
+ * Used to avoid duplication when API warnings are displayed separately.
+ * @param detail The detail string that may contain legacy warnings
+ * @returns The detail string with [Warning] sections removed, or null if only warnings remain
+ */
+export function stripWarningsFromDetail(detail: string | undefined): string | null {
+  if (!detail) {
+    return null;
+  }
+  LEGACY_WARNING_PATTERN.lastIndex = 0;
+  const stripped = detail.replace(LEGACY_WARNING_PATTERN, "").trim();
+  return stripped.length > 0 ? stripped : null;
+}

src/models/flinkStatement.ts

@@ -9,6 +9,8 @@ import type {
 } from "../clients/flinkSql";
 import { ConnectionType } from "../clients/sidecar";
 import { CCLOUD_BASE_PATH, CCLOUD_CONNECTION_ID, UTM_SOURCE_VSCODE } from "../constants";
+import type { StatementWarning } from "../flinkSql/warningParser";
+import { extractWarnings } from "../flinkSql/warningParser";
 import { IconNames } from "../icons";
 import type { IdItem } from "./main";
 import { CustomMarkdownString } from "./main";
@@ -266,6 +268,15 @@ export class FlinkStatement implements IResourceBase, IdItem, ISearchable, IEnvP
     return this.status?.detail;
   }
 
+  /**
+   * Get warnings for this statement.
+   * Prefers structured API warnings, falls back to parsing legacy [Warning] format from detail.
+   */
+  get warnings(): StatementWarning[] {
+    const statusWarnings = this.status.warnings;
+    return extractWarnings(statusWarnings, this.detail);
+  }
+
   get startTime(): Date | undefined {
     return this.metadata?.created_at;
   }