Add ClickHouse read/write nodes to the SQL graph

spark-ui/src/components/SqlFlow/MetricProcessors.tsx

@@ -335,6 +335,80 @@ export const processOutputNodeMetrics = (node: EnrichedSqlNode): MetricWithToolt
     return metrics;
 };
 
+// Renamed (pre-capitalization) ClickHouse write metric names that are collapsed into
+// the compact array rows below, so they can be filtered out of the raw metric list.
+const clickHouseBatchFillMetrics = [
+    "batches 0-25% full",
+    "batches 25-50% full",
+    "batches 50-75% full",
+    "batches 75-100% full",
+];
+const clickHouseBatchSizeMetrics = ["min batch size", "max batch size"];
+const clickHouseWriteCountMetrics = ["batch writes", "failed writes"];
+const clickHouseCollapsedMetrics = new Set([
+    ...clickHouseBatchFillMetrics,
+    ...clickHouseBatchSizeMetrics,
+    ...clickHouseWriteCountMetrics,
+]);
+
+// The verbose ClickHouse write metrics collapsed into compact array-style rows.
+// avg batch size is derived as rows / batch writes. Empty for non-ClickHouse nodes.
+const buildClickHouseWriteMetrics = (node: EnrichedSqlNode): MetricWithTooltip[] => {
+    const value = (name: string) => node.metrics.find((m) => m.name === name)?.value;
+    const num = (v: string | undefined) =>
+        v === undefined ? undefined : parseFloat(v.replaceAll(",", "").trim());
+    const cell = (v: string | number | undefined) =>
+        v === undefined ? "—" : v.toString().replaceAll(",", "").trim();
+    const array = (cells: (string | number | undefined)[]) =>
+        `[${cells.map(cell).join(", ")}]`;
+    const present = (cells: (string | undefined)[]) => cells.some((v) => v !== undefined);
+
+    const fill = clickHouseBatchFillMetrics.map(value);
+    const [minBatchSize, maxBatchSize] = clickHouseBatchSizeMetrics.map(value);
+    const [okWrites, failedWrites] = clickHouseWriteCountMetrics.map(value);
+
+    const rows = num(value("rows"));
+    const batchWrites = num(okWrites);
+    const avgBatchSize =
+        rows !== undefined && batchWrites ? Math.round(rows / batchWrites) : undefined;
+
+    const candidates: (MetricWithTooltip | false)[] = [
+        present(fill) && {
+            name: "Batch Fill (0-25/25-50/50-75/75-100%)",
+            value: array(fill),
+            tooltip:
+                "Number of write batches by how full they were, as a fraction of the configured batch size: 0-25%, 25-50%, 50-75%, 75-100%.",
+        },
+        present([minBatchSize, maxBatchSize]) && {
+            name: "Batch Size (min/avg/max)",
+            value: array([minBatchSize, avgBatchSize, maxBatchSize]),
+            tooltip:
+                "Rows per write batch: minimum, average (rows / batch writes), and maximum.",
+        },
+        present([okWrites, failedWrites]) && {
+            name: "Batch Writes (ok/failed)",
+            value: array([okWrites, failedWrites]),
+            tooltip: "Write batches sent to ClickHouse: successful and failed attempts.",
+        },
+    ];
+
+    return candidates.filter((m): m is MetricWithTooltip => m !== false);
+};
+
+// Replaces the individual ClickHouse write metric rows with the collapsed array rows.
+// Non-ClickHouse nodes are returned unchanged.
+export const collapseClickHouseWriteMetrics = (
+    node: EnrichedSqlNode,
+    baseMetrics: MetricWithTooltip[],
+): MetricWithTooltip[] => {
+    const collapsed = buildClickHouseWriteMetrics(node);
+    if (collapsed.length === 0) return baseMetrics;
+    return [
+        ...baseMetrics.filter((m) => !clickHouseCollapsedMetrics.has(m.name.toLowerCase())),
+        ...collapsed,
+    ];
+};
+
 export const processOptimizeTableMetrics = (node: EnrichedSqlNode): MetricWithTooltip[] => {
     const metrics: MetricWithTooltip[] = [];
 

spark-ui/src/components/SqlFlow/StageNode.tsx

@@ -13,6 +13,7 @@ import { TransperantTooltip } from "../AlertBadge/AlertBadge";
 import MetricDisplay, { MetricWithTooltip } from "./MetricDisplay";
 import {
   addTruncatedCodeTooltip,
+  collapseClickHouseWriteMetrics,
   processBaseMetrics,
   processCachedStorageMetrics,
   processDeltaLakeScanMetrics,
@@ -154,6 +155,10 @@ const StageNodeComponent: FC<StageNodeProps> = ({ data }) => {
       }
 
       const inputMetrics = processInputNodeMetrics(data.node);
+
+      // ClickHouse writes: collapse the verbose per-batch metrics into compact array rows.
+      allBaseMetrics = collapseClickHouseWriteMetrics(data.node, allBaseMetrics);
+
       metrics = [
         ...inputMetrics,
         ...allBaseMetrics,

spark-ui/src/interfaces/AppStore.ts

@@ -201,11 +201,17 @@ export type ParsedIcebergWritePlan = {
   tableType: string;
 };
 
+export type ParsedClickHouseWritePlan = {
+  database: string;
+  tableName: string;
+};
+
 export type ParseFileScanPlan = {
   Location?: string;
   tableName?: string;
   isIcebergRead?: boolean;
   isBigQueryRead?: boolean;
+  isClickHouseRead?: boolean;
   orderBy?: string[];
   PartitionFilters?: string[];
   PushedFilters?: string[];
@@ -322,7 +328,8 @@ export type ParsedNodePlan =
   | { type: "BatchEvalPython"; plan: ParsedBatchEvalPythonPlan }
   | { type: "Generate"; plan: ParsedGeneratePlan }
   | { type: "Expand"; plan: ParsedExpandPlan }
-  | { type: "WriteToIceberg"; plan: ParsedIcebergWritePlan };
+  | { type: "WriteToIceberg"; plan: ParsedIcebergWritePlan }
+  | { type: "WriteToClickHouse"; plan: ParsedClickHouseWritePlan };
 
 
 export interface ExchangeMetrics {

spark-ui/src/reducers/PlanParsers/ScanFileParser.spec.ts

@@ -116,6 +116,20 @@ describe("parseFileScan", () => {
         },
       },
     },
+    {
+      // A ClickHouse read looks like an Iceberg BatchScan by node name; the connector's
+      // scan class in the plan description disambiguates it (and avoids the Iceberg default).
+      input:
+        "BatchScan[id#16] class com.clickhouse.spark.read.ClickHouseBatchScan ReadSchema: struct<id:int>",
+      nodeName: "BatchScan `dataflint_demo`.`t`",
+      expected: {
+        tableName: "`dataflint_demo`.`t`",
+        isClickHouseRead: true,
+        ReadSchema: {
+          id: "int",
+        },
+      },
+    },
   ];
 
   testCases.forEach((testCase, idx) => {

spark-ui/src/reducers/PlanParsers/ScanFileParser.ts

@@ -80,7 +80,15 @@ export function parseFileScan(
       result.isIcebergRead = true;
     }
   } else if (nodeName.startsWith("BatchScan ")) {
-    if (readingTableMatch) {
+    if (input.includes("ClickHouseBatchScan")) {
+      // A ClickHouse read looks like an Iceberg scan by node name ("BatchScan `db`.`t`");
+      // the connector's scan class in the plan description is what disambiguates it.
+      const parts = nodeName.split(" ");
+      if (parts.length === 2) {
+        result.tableName = parts[1];
+      }
+      result.isClickHouseRead = true;
+    } else if (readingTableMatch) {
       result.tableName = readingTableMatch[1];
       result.isBigQueryRead = true;
     } else {

spark-ui/src/reducers/PlanParsers/WriteToClickHouseParser.spec.ts

@@ -0,0 +1,29 @@
+import { parseWriteToClickHouse } from "./WriteToClickHouseParser";
+
+describe("parseWriteToClickHouse", () => {
+  // What the deployed connector actually renders in the physical plan: the write
+  // object's default toString (class@hashcode), which carries no table info.
+  it("should return empty fields for the runtime class@hash form (detection still relies on the marker)", () => {
+    const input =
+      "AppendData Arguments: org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy$$Lambda$2639@21e38dcb, com.clickhouse.spark.write.ClickHouseWrite@671b3ff1";
+    expect(input.includes("ClickHouseWrite")).toBe(true);
+    expect(parseWriteToClickHouse(input)).toEqual({ database: "", tableName: "" });
+  });
+
+  // Forward-compatible: if a connector version prints Write.description instead.
+  it("should parse database and table when the description form is present", () => {
+    const input =
+      "AppendData Arguments: ClickHouseWrite(database=default, table=sales_by_store)})";
+    expect(parseWriteToClickHouse(input)).toEqual({
+      database: "default",
+      tableName: "sales_by_store",
+    });
+  });
+
+  it("should return empty strings when the description does not match", () => {
+    expect(parseWriteToClickHouse("AppendData something else")).toEqual({
+      database: "",
+      tableName: "",
+    });
+  });
+});

spark-ui/src/reducers/PlanParsers/WriteToClickHouseParser.ts

@@ -0,0 +1,12 @@
+import { ParsedClickHouseWritePlan } from "../../interfaces/AppStore";
+
+// The ClickHouse connector's Write.description renders as:
+//   ClickHouseWrite(database=<db>, table=<table>)})
+export function parseWriteToClickHouse(input: string): ParsedClickHouseWritePlan {
+  const databaseMatch = /ClickHouseWrite\(database=([^,]+),/.exec(input);
+  const tableMatch = /table=([^)]+)\)/.exec(input);
+  return {
+    database: databaseMatch?.[1]?.trim() ?? "",
+    tableName: tableMatch?.[1]?.trim() ?? "",
+  };
+}

spark-ui/src/reducers/SqlReducer.ts

@@ -38,6 +38,7 @@ import { parseSort } from "./PlanParsers/SortParser";
 import { parseTakeOrderedAndProject } from "./PlanParsers/TakeOrderedAndProjectParser";
 import { parseWindow } from "./PlanParsers/WindowParser";
 import { parseWriteToIceberg } from "./PlanParsers/WriteToIcebergParser";
+import { parseWriteToClickHouse } from "./PlanParsers/WriteToClickHouseParser";
 import { parseWriteToDelta } from "./PlanParsers/WriteToDeltaParser";
 import { parseWriteToHDFS } from "./PlanParsers/WriteToHDFSParser";
 import { parseBatchEvalPython } from "./PlanParsers/batchEvalPythonParser";
@@ -141,6 +142,12 @@ export function parseNodePlan(
             plan: parseWriteToIceberg(plan.planDescription),
           };
         }
+        if (plan.planDescription.includes("ClickHouseWrite")) {
+          return {
+            type: "WriteToClickHouse",
+            plan: parseWriteToClickHouse(plan.planDescription),
+          };
+        }
         break;
       case "Execute InsertIntoHadoopFsRelationCommand":
         return {

spark-ui/src/reducers/SqlReducerUtils.spec.ts

@@ -63,6 +63,51 @@ describe("nodeEnrichedNameBuilder - Iceberg write", () => {
   });
 });
 
+describe("nodeEnrichedNameBuilder - ClickHouse write", () => {
+  it("should return 'ClickHouse Write' for AppendData with a WriteToClickHouse plan", () => {
+    const plan: ParsedNodePlan = {
+      type: "WriteToClickHouse",
+      plan: { database: "default", tableName: "sales_by_store" },
+    };
+    expect(nodeEnrichedNameBuilder("AppendData", plan)).toBe("ClickHouse Write");
+  });
+
+  it("should fall back to the generic name without a WriteToClickHouse plan", () => {
+    expect(nodeEnrichedNameBuilder("AppendData", undefined)).toBe("Append data");
+  });
+});
+
+describe("calcNodeMetrics - ClickHouse write metrics", () => {
+  it("should keep and rename ClickHouse write metrics for output nodes", () => {
+    const metrics: EnrichedSqlMetric[] = [
+      { name: "number of output rows", value: "50" },
+      { name: "written output", value: "0.0 B" },
+      { name: "total batch writes to ClickHouse", value: "1" },
+      { name: "clients connected to ClickHouse", value: "1" },
+      { name: "failed write attempts to ClickHouse", value: "0" },
+      { name: "total time of writing", value: "0 ms" },
+      { name: "total time of serialization", value: "0 ms" },
+      { name: "max batch size written (rows)", value: "50" },
+      { name: "some unknown metric", value: "xyz" },
+    ];
+    const result = calcNodeMetrics("output", metrics);
+    const names = result.map((m) => m.name);
+
+    expect(names).toContain("rows");
+    expect(names).toContain("bytes written");
+    expect(names).toContain("batch writes");
+    expect(names).toContain("failed writes");
+    expect(names).toContain("write time");
+    expect(names).toContain("max batch size");
+    // removed from the node face
+    expect(names).not.toContain("ClickHouse clients");
+    expect(names).not.toContain("serialization time");
+    expect(names).not.toContain("some unknown metric");
+
+    expect(result.find((m) => m.name === "batch writes")?.value).toBe("1");
+  });
+});
+
 describe("nodeEnrichedNameBuilder - BigQuery", () => {
   it("should return 'BigQuery Read' for a FileScan plan with isBigQueryRead", () => {
     const plan: ParsedNodePlan = {
@@ -75,6 +120,16 @@ describe("nodeEnrichedNameBuilder - BigQuery", () => {
     expect(nodeEnrichedNameBuilder("BatchScan myproject.mydataset.mytable", plan)).toBe("BigQuery Read");
   });
 
+  it("should return 'ClickHouse Read' for a FileScan plan with isClickHouseRead", () => {
+    const plan: ParsedNodePlan = {
+      type: "FileScan",
+      plan: { tableName: "`dataflint_demo`.`t`", isClickHouseRead: true },
+    };
+    expect(nodeEnrichedNameBuilder("BatchScan `dataflint_demo`.`t`", plan)).toBe(
+      "ClickHouse Read",
+    );
+  });
+
   it("should NOT return 'BigQuery Read' for a FileScan plan without isBigQueryRead", () => {
     const plan: ParsedNodePlan = {
       type: "FileScan",

spark-ui/src/reducers/SqlReducerUtils.ts

@@ -35,6 +35,16 @@ const metricAllowlist: Record<NodeType, Array<string>> = {
     "total number of files merged by ZOrderBy",
     "total bytes in files merged by ZOrderBy",
     "duration",
+    // ClickHouse connector write metrics
+    "total batch writes to ClickHouse",
+    "failed write attempts to ClickHouse",
+    "total time of writing",
+    "min batch size written (rows)",
+    "max batch size written (rows)",
+    "batches 0-25% of configured batch size",
+    "batches 25-50% of configured batch size",
+    "batches 50-75% of configured batch size",
+    "batches 75-100% of configured batch size",
   ],
   join: [
     "number of output rows",
@@ -194,6 +204,16 @@ const metricsRenamer: Record<string, string> = {
   "time of hash probe": "hash probe time",
   "number of spilled bytes": "spill",
   "peak memory bytes": "peak memory",
+  // ClickHouse connector write metrics
+  "total batch writes to ClickHouse": "batch writes",
+  "failed write attempts to ClickHouse": "failed writes",
+  "total time of writing": "write time",
+  "min batch size written (rows)": "min batch size",
+  "max batch size written (rows)": "max batch size",
+  "batches 0-25% of configured batch size": "batches 0-25% full",
+  "batches 25-50% of configured batch size": "batches 25-50% full",
+  "batches 50-75% of configured batch size": "batches 50-75% full",
+  "batches 75-100% of configured batch size": "batches 75-100% full",
 };
 
 const nodeTypeDict: Record<string, NodeType> = {
@@ -470,6 +490,7 @@ export function nodeEnrichedNameBuilder(
     switch (plan.type) {
       case "FileScan":
         if (plan.plan.isBigQueryRead) return "BigQuery Read";
+        if (plan.plan.isClickHouseRead) return "ClickHouse Read";
         break;
       case "WriteToIceberg":
         if (name === "OverwriteByExpression") return "Iceberg - Overwrite by Expression";
@@ -478,6 +499,8 @@ export function nodeEnrichedNameBuilder(
         if (name === "WriteDelta") return "Iceberg - Write Delta";
         if (name === "DeleteFromTable") return "Iceberg - Delete from table";
         return "Iceberg - Append data";
+      case "WriteToClickHouse":
+        return "ClickHouse Write";
       case "JDBCScan":
         return "Read JDBC";
       case "HashAggregate":

spark-ui/src/utils/FormatUtils.ts

@@ -14,7 +14,15 @@ export function humanFileSize(bytes: number): string {
 export function capitalizeWords(text: string): string {
   return text
     .split(' ')
-    .map(word => word.charAt(0).toUpperCase() + word.slice(1).toLowerCase())
+    .map(word => {
+      // Preserve words that already carry internal casing (acronyms and
+      // camelCase brand names like "HDFS", "BigQuery", "ClickHouse") instead
+      // of lower-casing them into "Hdfs" / "Bigquery" / "Clickhouse".
+      if (/[A-Z]/.test(word.slice(1))) {
+        return word;
+      }
+      return word.charAt(0).toUpperCase() + word.slice(1).toLowerCase();
+    })
     .join(' ');
 }