From 32a12187c3bf67236bb18bf64ecb6cd0eb3e6bc4 Mon Sep 17 00:00:00 2001 From: marais Date: Fri, 10 Jul 2026 16:13:56 +0300 Subject: [PATCH] Add ClickHouse read/write nodes to the SQL graph Recognise the ClickHouse Spark connector in the query plan and give its nodes first-class treatment in the DataFlint SQL view: - Writes: an AppendData whose plan references ClickHouseWrite is renamed "ClickHouse Write" and surfaces the connector's write metrics. - Reads: a BatchScan backed by ClickHouseBatchScan is renamed "ClickHouse Read". Previously these were misdetected as Iceberg reads, since they look identical by node name; the scan class in the plan description disambiguates them. - The verbose per-batch write metrics are collapsed into compact array-style rows so they fit on the node: Batch Fill (0-25/25-50/50-75/75-100%): [a, b, c, d] Batch Size (min/avg/max): [min, avg, max] (avg = rows / batch writes) Batch Writes (ok/failed): [ok, failed] - capitalizeWords now preserves mixed-case brand/acronym words, so node names render as "ClickHouse"/"BigQuery"/"HDFS" instead of "Clickhouse"/"Bigquery"/"Hdfs". Co-Authored-By: Claude Opus 4.8 (1M context) --- .../components/SqlFlow/MetricProcessors.tsx | 74 +++++++++++++++++++ spark-ui/src/components/SqlFlow/StageNode.tsx | 5 ++ spark-ui/src/interfaces/AppStore.ts | 9 ++- .../PlanParsers/ScanFileParser.spec.ts | 14 ++++ .../reducers/PlanParsers/ScanFileParser.ts | 10 ++- .../WriteToClickHouseParser.spec.ts | 29 ++++++++ .../PlanParsers/WriteToClickHouseParser.ts | 12 +++ spark-ui/src/reducers/SqlReducer.ts | 7 ++ spark-ui/src/reducers/SqlReducerUtils.spec.ts | 55 ++++++++++++++ spark-ui/src/reducers/SqlReducerUtils.ts | 23 ++++++ spark-ui/src/utils/FormatUtils.ts | 10 ++- 11 files changed, 245 insertions(+), 3 deletions(-) create mode 100644 spark-ui/src/reducers/PlanParsers/WriteToClickHouseParser.spec.ts create mode 100644 spark-ui/src/reducers/PlanParsers/WriteToClickHouseParser.ts diff --git a/spark-ui/src/components/SqlFlow/MetricProcessors.tsx b/spark-ui/src/components/SqlFlow/MetricProcessors.tsx index e4cad8e4..4e2cf1aa 100644 --- a/spark-ui/src/components/SqlFlow/MetricProcessors.tsx +++ b/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[] = []; diff --git a/spark-ui/src/components/SqlFlow/StageNode.tsx b/spark-ui/src/components/SqlFlow/StageNode.tsx index c8901630..2effd64a 100644 --- a/spark-ui/src/components/SqlFlow/StageNode.tsx +++ b/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 = ({ 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, diff --git a/spark-ui/src/interfaces/AppStore.ts b/spark-ui/src/interfaces/AppStore.ts index af056548..833d41e6 100644 --- a/spark-ui/src/interfaces/AppStore.ts +++ b/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 { diff --git a/spark-ui/src/reducers/PlanParsers/ScanFileParser.spec.ts b/spark-ui/src/reducers/PlanParsers/ScanFileParser.spec.ts index 6d467151..1f0a9760 100644 --- a/spark-ui/src/reducers/PlanParsers/ScanFileParser.spec.ts +++ b/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", + nodeName: "BatchScan `dataflint_demo`.`t`", + expected: { + tableName: "`dataflint_demo`.`t`", + isClickHouseRead: true, + ReadSchema: { + id: "int", + }, + }, + }, ]; testCases.forEach((testCase, idx) => { diff --git a/spark-ui/src/reducers/PlanParsers/ScanFileParser.ts b/spark-ui/src/reducers/PlanParsers/ScanFileParser.ts index e404e555..64a3b8ca 100644 --- a/spark-ui/src/reducers/PlanParsers/ScanFileParser.ts +++ b/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 { diff --git a/spark-ui/src/reducers/PlanParsers/WriteToClickHouseParser.spec.ts b/spark-ui/src/reducers/PlanParsers/WriteToClickHouseParser.spec.ts new file mode 100644 index 00000000..59143c31 --- /dev/null +++ b/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: "", + }); + }); +}); diff --git a/spark-ui/src/reducers/PlanParsers/WriteToClickHouseParser.ts b/spark-ui/src/reducers/PlanParsers/WriteToClickHouseParser.ts new file mode 100644 index 00000000..f58b0997 --- /dev/null +++ b/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=, 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() ?? "", + }; +} diff --git a/spark-ui/src/reducers/SqlReducer.ts b/spark-ui/src/reducers/SqlReducer.ts index 9ade387f..e4e573f8 100644 --- a/spark-ui/src/reducers/SqlReducer.ts +++ b/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 { diff --git a/spark-ui/src/reducers/SqlReducerUtils.spec.ts b/spark-ui/src/reducers/SqlReducerUtils.spec.ts index 504971f9..79174b4a 100644 --- a/spark-ui/src/reducers/SqlReducerUtils.spec.ts +++ b/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", diff --git a/spark-ui/src/reducers/SqlReducerUtils.ts b/spark-ui/src/reducers/SqlReducerUtils.ts index cc30347e..68284b6e 100644 --- a/spark-ui/src/reducers/SqlReducerUtils.ts +++ b/spark-ui/src/reducers/SqlReducerUtils.ts @@ -35,6 +35,16 @@ const metricAllowlist: Record> = { "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 = { "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 = { @@ -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": diff --git a/spark-ui/src/utils/FormatUtils.ts b/spark-ui/src/utils/FormatUtils.ts index 6d5f6a6b..cc99371c 100644 --- a/spark-ui/src/utils/FormatUtils.ts +++ b/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(' '); }