diff --git a/agent-service/src/agent/texera-agent.ts b/agent-service/src/agent/texera-agent.ts index ccd0545919a..8a2eaec273c 100644 --- a/agent-service/src/agent/texera-agent.ts +++ b/agent-service/src/agent/texera-agent.ts @@ -718,7 +718,7 @@ export class TexeraAgent { const result = new Map(); const visible = this.workflowResultState.getAllVisible(); for (const [operatorId, entry] of visible) { - result.set(operatorId, formatOperatorResult(operatorId, entry.operatorInfo, this.workflowState)); + result.set(operatorId, formatOperatorResult(operatorId, entry.operatorInfo)); } return result; } diff --git a/agent-service/src/agent/tools/result-formatting.spec.ts b/agent-service/src/agent/tools/result-formatting.spec.ts index e6d1afdf2e3..113cf99763b 100644 --- a/agent-service/src/agent/tools/result-formatting.spec.ts +++ b/agent-service/src/agent/tools/result-formatting.spec.ts @@ -19,104 +19,129 @@ import { describe, expect, test } from "bun:test"; import { formatOperatorResult } from "./result-formatting"; -import { WorkflowState } from "../workflow-state"; -import type { OperatorInfo } from "../../types/execution"; -import type { OperatorPredicate, OperatorLink, PortDescription } from "../../types/workflow"; - -function makeOpInfo(overrides: Partial = {}): OperatorInfo { +import { + ConsoleMessageType, + OperatorState, + OperatorResultMode, + WorkflowFatalErrorType, + type OperatorExecutionSummary, + type WorkflowFatalError, + type Tuple, +} from "../../types/execution"; + +// Build an engine-style Tuple with an all-STRING schema from a column->value record, +// matching how the backend emits truncated sampled rows. +function recordToTuple(row: Record): Tuple { return { - state: "completed", - inputTuples: 0, - outputTuples: 0, - resultMode: "table", - ...overrides, + schema: { attributes: Object.keys(row).map(name => ({ attributeName: name, attributeType: "string" })) }, + fields: Object.values(row), }; } -function makeOperator(id: string, inputPortIDs: string[] = []): OperatorPredicate { - const inputPorts: PortDescription[] = inputPortIDs.map((portID, i) => ({ - portID, - displayName: `Input ${i}`, - })); - return { - operatorID: id, - operatorType: "TestOp", - operatorVersion: "1.0", - operatorProperties: {}, - inputPorts, - outputPorts: [{ portID: "output-0", displayName: "Output 0" }], - showAdvanced: false, - }; +function toSampleRows(rows: Record[]): [number, Tuple][] { + return rows.map((row, rowIndex) => [rowIndex, recordToTuple(row)]); } -function makeLink(linkID: string, source: [string, string], target: [string, string]): OperatorLink { +interface OpInfoOverrides { + state?: OperatorState; + error?: string; + outputTuples?: number; + totalTuplesCount?: number; + warnings?: string[]; + result?: Record[]; + sampleTuples?: [number, Tuple][]; + resultMode?: OperatorResultMode; +} + +function makeExecutionFailure(message: string): WorkflowFatalError { return { - linkID, - source: { operatorID: source[0], portID: source[1] }, - target: { operatorID: target[0], portID: target[1] }, + type: { name: WorkflowFatalErrorType.EXECUTION_FAILURE }, + timestamp: { seconds: 0, nanos: 0 }, + message, + details: "", + operatorId: "", + workerId: "", }; } -const EMPTY_STATE = new WorkflowState(); +function makeOpInfo(overrides: OpInfoOverrides = {}): OperatorExecutionSummary { + const summary: OperatorExecutionSummary = { + state: overrides.state ?? OperatorState.COMPLETED, + errorMessages: overrides.error ? [makeExecutionFailure(overrides.error)] : [], + }; + // The result summary is present only when the operator produced a result. + if (overrides.result !== undefined || overrides.sampleTuples !== undefined) { + summary.resultSummary = { + resultMode: overrides.resultMode ?? OperatorResultMode.TABLE, + // Non-arrays are passed through to exercise the "(no result data)" guard. + sampleTuples: + overrides.sampleTuples ?? + (Array.isArray(overrides.result) ? toSampleRows(overrides.result) : (overrides.result as any)), + totalTuplesCount: overrides.totalTuplesCount ?? overrides.outputTuples ?? 0, + }; + } + if (overrides.warnings) { + // Warnings are derived from console messages whose title is "WARNING: ...". + summary.consoleMessages = overrides.warnings.map(w => ({ + msgType: ConsoleMessageType.PRINT, + title: w, + message: "", + })); + } + return summary; +} describe("formatOperatorResult - early returns", () => { test("returns [ERROR] prefix when error field is set", () => { - const out = formatOperatorResult("op1", makeOpInfo({ error: "boom" }), EMPTY_STATE); + const out = formatOperatorResult("op1", makeOpInfo({ error: "boom" })); expect(out).toBe("[ERROR] boom"); }); test("treats empty-string error as falsy and continues to result path", () => { - const out = formatOperatorResult("op1", makeOpInfo({ error: "" }), EMPTY_STATE); + const out = formatOperatorResult("op1", makeOpInfo({ error: "" })); expect(out).not.toContain("[ERROR]"); expect(out).toContain("(no result data)"); }); test("returns (no result data) when result is undefined", () => { - const out = formatOperatorResult("op1", makeOpInfo(), EMPTY_STATE); + const out = formatOperatorResult("op1", makeOpInfo()); expect(out).toBe("(no result data)"); }); test("returns (no result data) when result is not an array", () => { - const out = formatOperatorResult( - "op1", - makeOpInfo({ result: { rows: [] } as unknown as Record[] }), - EMPTY_STATE - ); + const out = formatOperatorResult("op1", makeOpInfo({ result: { rows: [] } as unknown as Record[] })); expect(out).toBe("(no result data)"); }); test("empty array result emits brief summary plus zero-column shape only", () => { - const out = formatOperatorResult("op1", makeOpInfo({ result: [], outputTuples: 0 }), EMPTY_STATE); + const out = formatOperatorResult("op1", makeOpInfo({ result: [], outputTuples: 0 })); expect(out.split("\n")).toEqual(["Executed operator op1", "Output table shape: (0, 0)"]); }); }); describe("formatOperatorResult - table shape and metadata", () => { - test("uses outputTuples for row count when totalRowCount missing", () => { - const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 7, result: [{ a: 1, b: 2 }] }), EMPTY_STATE); + test("uses outputTuples for row count when totalTuplesCount missing", () => { + const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 7, result: [{ a: 1, b: 2 }] })); expect(out).toContain("Output table shape: (7, 2)"); }); - test("totalRowCount overrides outputTuples in output shape", () => { + test("totalTuplesCount overrides outputTuples in output shape", () => { const out = formatOperatorResult( "op1", - makeOpInfo({ outputTuples: 7, totalRowCount: 999, result: [{ a: 1, b: 2 }] }), - EMPTY_STATE + makeOpInfo({ outputTuples: 7, totalTuplesCount: 999, result: [{ a: 1, b: 2 }] }) ); expect(out).toContain("Output table shape: (999, 2)"); }); - test("filters internal __is_visualization__ key from outer column count", () => { + test("counts every result tuple key as a column", () => { const out = formatOperatorResult( "op1", makeOpInfo({ outputTuples: 1, - result: [{ __is_visualization__: true, "html-content": "" }], - }), - EMPTY_STATE + result: [{ "html-content": "", label: "chart" }], + }) ); - // 1 visible column ("html-content") since __is_visualization__ is filtered. - expect(out).toContain("Output table shape: (1, 1)"); + expect(out).toContain("Output table shape: (1, 2)"); }); test("appends warnings after metadata lines", () => { @@ -125,104 +150,32 @@ describe("formatOperatorResult - table shape and metadata", () => { makeOpInfo({ outputTuples: 1, result: [{ a: 1 }], - warnings: ["truncated to 1 row", "something else"], - }), - EMPTY_STATE + warnings: ["WARNING: truncated to 1 row", "WARNING: something else"], + }) ); const lines = out.split("\n"); expect(lines[0]).toBe("Executed operator op1"); expect(lines[1]).toBe("Output table shape: (1, 1)"); - expect(lines[2]).toBe("truncated to 1 row"); - expect(lines[3]).toBe("something else"); - }); -}); - -describe("formatOperatorResult - input port metadata", () => { - test("omits input metadata when inputPortShapes is missing", () => { - const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 1, result: [{ a: 1 }] }), EMPTY_STATE); - expect(out).not.toContain("Input operator"); - }); - - test("omits input metadata when inputPortShapes is empty", () => { - const out = formatOperatorResult( - "op1", - makeOpInfo({ outputTuples: 1, result: [{ a: 1 }], inputPortShapes: [] }), - EMPTY_STATE - ); - expect(out).not.toContain("Input operator"); - }); - - test("falls back to inputN placeholder when no upstream link matches the port", () => { - const out = formatOperatorResult( - "op1", - makeOpInfo({ - outputTuples: 1, - result: [{ a: 1 }], - inputPortShapes: [{ portIndex: 0, rows: 5, columns: 3 }], - }), - EMPTY_STATE - ); - expect(out).toContain("Input operator(table shape): input0(5, 3)"); - }); - - test("uses upstream operator id when an input link matches the port", () => { - const state = new WorkflowState(); - state.addOperator(makeOperator("upstream")); - state.addOperator(makeOperator("op1", ["input-0"])); - state.addLink(makeLink("l1", ["upstream", "output-0"], ["op1", "input-0"])); - - const out = formatOperatorResult( - "op1", - makeOpInfo({ - outputTuples: 4, - result: [{ a: 1, b: 2 }], - inputPortShapes: [{ portIndex: 0, rows: 10, columns: 2 }], - }), - state - ); - expect(out).toContain("Input operator(table shape): upstream(10, 2)"); - }); - - test("sorts multiple input ports by portIndex regardless of input order", () => { - const state = new WorkflowState(); - state.addOperator(makeOperator("up0")); - state.addOperator(makeOperator("up1")); - state.addOperator(makeOperator("op1", ["input-0", "input-1"])); - state.addLink(makeLink("l0", ["up0", "output-0"], ["op1", "input-0"])); - state.addLink(makeLink("l1", ["up1", "output-0"], ["op1", "input-1"])); - - const out = formatOperatorResult( - "op1", - makeOpInfo({ - outputTuples: 1, - result: [{ a: 1 }], - inputPortShapes: [ - { portIndex: 1, rows: 2, columns: 2 }, - { portIndex: 0, rows: 1, columns: 1 }, - ], - }), - state - ); - expect(out).toContain("Input operator(table shape): up0(1, 1), up1(2, 2)"); + expect(lines[2]).toBe("WARNING: truncated to 1 row"); + expect(lines[3]).toBe("WARNING: something else"); }); }); describe("formatOperatorResult - visualization rows", () => { - test("strips html-content and json-content payloads when row is flagged as visualization", () => { + test("strips html-content and json-content payloads when result mode is visualization", () => { const out = formatOperatorResult( "op1", makeOpInfo({ outputTuples: 1, + resultMode: OperatorResultMode.VISUALIZATION, result: [ { - __is_visualization__: true, "html-content": "
hidden
", "json-content": '{"big":1}', label: "chart", }, ], - }), - EMPTY_STATE + }) ); expect(out).toContain(""); expect(out).not.toContain("
hidden
"); @@ -230,40 +183,38 @@ describe("formatOperatorResult - visualization rows", () => { expect(out).toContain("chart"); }); - test("__is_visualization__ false leaves the visualization-only fields untouched", () => { + test("table result mode leaves visualization payload fields untouched", () => { const out = formatOperatorResult( "op1", makeOpInfo({ outputTuples: 1, - result: [{ __is_visualization__: false, "html-content": "" }], - }), - EMPTY_STATE + resultMode: OperatorResultMode.TABLE, + result: [{ "html-content": "" }], + }) ); expect(out).toContain(""); expect(out).not.toContain(""); }); - test("__is_visualization__ column is excluded from rendered table body and shape agrees", () => { + test("table rows render all tuple columns and shape agrees", () => { const out = formatOperatorResult( "op1", makeOpInfo({ outputTuples: 1, - result: [{ __is_visualization__: false, value: 1 }], - }), - EMPTY_STATE + result: [{ value: 1 }], + }) ); const lines = out.split("\n"); expect(out).toContain("Output table shape: (1, 1)"); // Header line is the third line (after brief summary and shape line). expect(lines[2]).toBe("\tvalue"); expect(lines[3]).toBe("0\t1"); - expect(out).not.toContain("__is_visualization__"); }); }); describe("jsonToTableFormat - cell coercion via formatOperatorResult", () => { - function tableLines(opInfo: Partial): string[] { - const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 1, ...opInfo }), EMPTY_STATE); + function tableLines(opInfo: OpInfoOverrides): string[] { + const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 1, ...opInfo })); // Skip brief summary + shape line. return out.split("\n").slice(2); } @@ -296,17 +247,16 @@ describe("jsonToTableFormat - cell coercion via formatOperatorResult", () => { }); describe("jsonToTableFormat - row index gaps", () => { - test("inserts ... separator when __row_index__ skips ahead", () => { + test("inserts ... separator when rowIndex skips ahead", () => { const out = formatOperatorResult( "op1", makeOpInfo({ outputTuples: 2, - result: [ - { __row_index__: 0, v: "a" }, - { __row_index__: 5, v: "b" }, + sampleTuples: [ + [0, recordToTuple({ v: "a" })], + [5, recordToTuple({ v: "b" })], ], - }), - EMPTY_STATE + }) ); const lines = out.split("\n"); // header, row0, gap marker, row5 @@ -316,26 +266,24 @@ describe("jsonToTableFormat - row index gaps", () => { expect(lines[lines.length - 1]).toBe("5\tb"); }); - test("no separator is emitted between consecutive __row_index__ values", () => { + test("no separator is emitted between consecutive rowIndex values", () => { const out = formatOperatorResult( "op1", makeOpInfo({ outputTuples: 2, - result: [ - { __row_index__: 0, v: "a" }, - { __row_index__: 1, v: "b" }, + sampleTuples: [ + [0, recordToTuple({ v: "a" })], + [1, recordToTuple({ v: "b" })], ], - }), - EMPTY_STATE + }) ); expect(out).not.toContain("...\t..."); }); - test("non-zero starting __row_index__ does not emit a leading gap marker", () => { + test("non-zero starting rowIndex does not emit a leading gap marker", () => { const out = formatOperatorResult( "op1", - makeOpInfo({ outputTuples: 1, result: [{ __row_index__: 9, v: "z" }] }), - EMPTY_STATE + makeOpInfo({ outputTuples: 1, sampleTuples: [[9, recordToTuple({ v: "z" })]] }) ); expect(out).not.toContain("...\t..."); expect(out.endsWith("9\tz")).toBe(true); diff --git a/agent-service/src/agent/tools/result-formatting.ts b/agent-service/src/agent/tools/result-formatting.ts index 5ed4aacc5d4..712aff68d1c 100644 --- a/agent-service/src/agent/tools/result-formatting.ts +++ b/agent-service/src/agent/tools/result-formatting.ts @@ -17,119 +17,50 @@ * under the License. */ -import type { OperatorInfo } from "../../types/execution"; -import type { WorkflowState } from "../workflow-state"; -import { formatExecuteOperatorResult, getVisibleResultHeaders } from "./tools-utility"; - -export function formatOperatorResult(operatorId: string, opInfo: OperatorInfo, workflowState: WorkflowState): string { - if (opInfo.error) { - return `[ERROR] ${opInfo.error}`; +import { OperatorResultMode, type OperatorExecutionSummary, type Tuple } from "../../types/execution"; +import { + formatExecuteOperatorResult, + formatSampleRowsAsTsv, + getOperatorErrorText, + getOperatorWarnings, + tupleColumns, +} from "./tools-utility"; + +export function formatOperatorResult(operatorId: string, opInfo: OperatorExecutionSummary): string { + const errorText = getOperatorErrorText(opInfo); + if (errorText) { + return `[ERROR] ${errorText}`; } - if (!opInfo.result || !Array.isArray(opInfo.result)) { + const sampleTuples = opInfo.resultSummary?.sampleTuples; + if (!sampleTuples || !Array.isArray(sampleTuples)) { return "(no result data)"; } - const jsonArray = opInfo.result as Record[]; - const headers = jsonArray.length > 0 ? getVisibleResultHeaders(jsonArray[0]) : []; - const columns = headers.length; - - const isViz = jsonArray.length > 0 && jsonArray[0]["__is_visualization__"] === true; - const serializableArray = isViz - ? jsonArray.map(row => { - const cleaned: Record = {}; - for (const key of Object.keys(row)) { - if (key === "__is_visualization__") continue; - if (key === "html-content" || key === "json-content") { - cleaned[key] = ""; - } else { - cleaned[key] = row[key]; - } - } - return cleaned; + const isViz = opInfo.resultSummary?.resultMode === OperatorResultMode.VISUALIZATION; + const rows: [number, Tuple][] = isViz + ? sampleTuples.map(([rowIndex, tuple]) => { + const fields = tuple.schema.attributes.map((a, i) => + a.attributeName === "html-content" || a.attributeName === "json-content" + ? "" + : tuple.fields[i] + ); + return [rowIndex, { schema: tuple.schema, fields }]; }) - : jsonArray; + : sampleTuples; - const dataString = jsonToTableFormat(serializableArray); + const headers = rows.length > 0 ? tupleColumns(rows[0][1]) : []; + const columns = headers.length; + + const dataString = formatSampleRowsAsTsv(rows); - const metadataLines = [ - formatInputOutputMetadata(workflowState, operatorId, opInfo, columns), - ...(opInfo.warnings ?? []), - ].filter(Boolean); + // Output shape only; input-port shapes are derivable by the agent from the DAG + // links plus each upstream operator's own output shape shown in context. + const outputRows = opInfo.resultSummary?.totalTuplesCount ?? 0; + const metadataLines = [`Output table shape: (${outputRows}, ${columns})`, ...getOperatorWarnings(opInfo)].filter( + Boolean + ); const briefSummary = formatExecuteOperatorResult(operatorId); return [briefSummary, ...metadataLines, dataString].filter(Boolean).join("\n"); } - -function formatInputOutputMetadata( - workflowState: WorkflowState, - operatorId: string, - opInfo: OperatorInfo, - outputColumns: number -): string { - const outputRows = opInfo.totalRowCount ?? opInfo.outputTuples; - const outputLine = `Output table shape: (${outputRows}, ${outputColumns})`; - - const inputShapes = opInfo.inputPortShapes; - if (!inputShapes || inputShapes.length === 0) { - return outputLine; - } - - const inputLinks = workflowState.getAllLinks().filter(l => l.target.operatorID === operatorId); - const portIndexToUpstream = new Map(); - const op = workflowState.getOperator(operatorId); - for (const link of inputLinks) { - const portIdx = op?.inputPorts.findIndex(p => p.portID === link.target.portID) ?? -1; - if (portIdx >= 0) { - portIndexToUpstream.set(portIdx, link.source.operatorID); - } - } - - const inputPart = inputShapes - .sort((a, b) => a.portIndex - b.portIndex) - .map(p => { - const name = portIndexToUpstream.get(p.portIndex) ?? `input${p.portIndex}`; - return `${name}(${p.rows}, ${p.columns})`; - }) - .join(", "); - - return `Input operator(table shape): ${inputPart}\n${outputLine}`; -} - -function jsonToTableFormat(jsonResult: Record[]): string { - if (!jsonResult || jsonResult.length === 0) return ""; - - const hasRowIndex = "__row_index__" in jsonResult[0]; - const headers = getVisibleResultHeaders(jsonResult[0]); - if (headers.length === 0) return ""; - - const headerLine = "\t" + headers.join("\t"); - const formattedRows: string[] = []; - let prevIndex = -1; - - for (let i = 0; i < jsonResult.length; i++) { - const row = jsonResult[i]; - const rowIndex = hasRowIndex ? (row["__row_index__"] as number) : i; - - if (prevIndex >= 0 && rowIndex > prevIndex + 1) { - const dots = headers.map(() => "...").join("\t"); - formattedRows.push(`...\t${dots}`); - } - prevIndex = rowIndex; - - const cells = headers.map(h => { - const val = row[h]; - if (val === null) return "NaN"; - if (val === undefined) return ""; - if (typeof val === "number" || typeof val === "boolean") return String(val); - if (typeof val === "string") { - if (val === "NULL") return "NaN"; - return val.replace(/\t/g, "\\t").replace(/\n/g, "\\n"); - } - return JSON.stringify(val); - }); - formattedRows.push(`${rowIndex}\t${cells.join("\t")}`); - } - - return [headerLine, ...formattedRows].join("\n"); -} diff --git a/agent-service/src/agent/tools/tools-utility.spec.ts b/agent-service/src/agent/tools/tools-utility.spec.ts index b505199709e..dc612e5243a 100644 --- a/agent-service/src/agent/tools/tools-utility.spec.ts +++ b/agent-service/src/agent/tools/tools-utility.spec.ts @@ -25,40 +25,66 @@ import { formatModifyOperatorResult, formatExecuteOperatorResult, formatOperatorError, - getVisibleResultHeaders, + getOperatorErrorText, + tupleColumns, + tupleToRecord, } from "./tools-utility"; +import { OperatorState, WorkflowFatalErrorType, type WorkflowFatalError, type Tuple } from "../../types/execution"; -describe("getVisibleResultHeaders", () => { - test("returns every key when no internal columns are present", () => { - expect(getVisibleResultHeaders({ a: 1, b: 2 })).toEqual(["a", "b"]); +function makeFatal(message: string): WorkflowFatalError { + return { + type: { name: WorkflowFatalErrorType.EXECUTION_FAILURE }, + timestamp: { seconds: 0, nanos: 0 }, + message, + details: "", + operatorId: "", + workerId: "", + }; +} + +describe("getOperatorErrorText", () => { + test("joins the messages of every fatal error", () => { + const opInfo = { state: OperatorState.FAILED, errorMessages: [makeFatal("e1"), makeFatal("e2")] }; + expect(getOperatorErrorText(opInfo)).toBe("e1; e2"); }); - test("strips __row_index__ from the result", () => { - expect(getVisibleResultHeaders({ __row_index__: 0, a: 1 })).toEqual(["a"]); + test("ignores errors whose message is empty", () => { + const opInfo = { state: OperatorState.FAILED, errorMessages: [makeFatal(""), makeFatal("real")] }; + expect(getOperatorErrorText(opInfo)).toBe("real"); }); - test("strips __is_visualization__ from the result", () => { - expect(getVisibleResultHeaders({ __is_visualization__: true, a: 1 })).toEqual(["a"]); + test("returns an empty string when every error message is empty", () => { + const opInfo = { state: OperatorState.COMPLETED, errorMessages: [makeFatal("")] }; + expect(getOperatorErrorText(opInfo)).toBe(""); }); - test("strips every known internal column at once", () => { - expect(getVisibleResultHeaders({ __row_index__: 0, __is_visualization__: true, a: 1, b: 2 })).toEqual(["a", "b"]); + test("returns an empty string when there are no errors", () => { + const opInfo = { state: OperatorState.COMPLETED, errorMessages: [] }; + expect(getOperatorErrorText(opInfo)).toBe(""); }); +}); - test("preserves visible column order", () => { - expect(getVisibleResultHeaders({ z: 1, __row_index__: 0, a: 2, __is_visualization__: true, m: 3 })).toEqual([ - "z", - "a", - "m", - ]); +describe("tupleColumns / tupleToRecord", () => { + const tuple: Tuple = { + schema: { + attributes: [ + { attributeName: "z", attributeType: "string" }, + { attributeName: "a", attributeType: "string" }, + ], + }, + fields: [1, 2], + }; + + test("tupleColumns returns column names in schema order", () => { + expect(tupleColumns(tuple)).toEqual(["z", "a"]); }); - test("returns an empty array for an empty row", () => { - expect(getVisibleResultHeaders({})).toEqual([]); + test("tupleToRecord projects fields back onto their column names", () => { + expect(tupleToRecord(tuple)).toEqual({ z: 1, a: 2 }); }); - test("returns an empty array when only internal columns are present", () => { - expect(getVisibleResultHeaders({ __row_index__: 0, __is_visualization__: true })).toEqual([]); + test("tupleColumns is empty for a schema with no attributes", () => { + expect(tupleColumns({ schema: { attributes: [] }, fields: [] })).toEqual([]); }); }); diff --git a/agent-service/src/agent/tools/tools-utility.ts b/agent-service/src/agent/tools/tools-utility.ts index 6c9ab004f6e..1c44e01f48b 100644 --- a/agent-service/src/agent/tools/tools-utility.ts +++ b/agent-service/src/agent/tools/tools-utility.ts @@ -17,10 +17,72 @@ * under the License. */ -export const INTERNAL_RESULT_KEYS: ReadonlySet = new Set(["__row_index__", "__is_visualization__"]); +import type { OperatorExecutionSummary, Tuple } from "../../types/execution"; -export function getVisibleResultHeaders(row: Record): string[] { - return Object.keys(row).filter(k => !INTERNAL_RESULT_KEYS.has(k)); +// The single definition of "this operator failed": some fatal error carries +// message text. The engine can emit console ERRORs with empty text, which do +// not count, matching the previous `error` field's truthiness semantics. +export function getOperatorErrorText(opInfo: OperatorExecutionSummary): string { + return opInfo.errorMessages + .map(e => e.message) + .filter(Boolean) + .join("; "); +} + +// The column names of a tuple, in schema order. +export function tupleColumns(tuple: Tuple): string[] { + return tuple.schema.attributes.map(a => a.attributeName); +} + +// Project a tuple's positional fields back into a column->value record. +export function tupleToRecord(tuple: Tuple): Record { + const record: Record = {}; + tuple.schema.attributes.forEach((a, i) => { + record[a.attributeName] = tuple.fields[i]; + }); + return record; +} + +// Sampled rows arrive as [originalRowIndex, Tuple] pairs. +export function formatSampleRowsAsTsv(rows: [number, Tuple][]): string { + if (!rows || rows.length === 0) return ""; + + const headers = tupleColumns(rows[0][1]); + if (headers.length === 0) return ""; + + const headerLine = "\t" + headers.join("\t"); + const formattedRows: string[] = []; + let prevIndex = -1; + + for (const [rowIndex, tuple] of rows) { + if (prevIndex >= 0 && rowIndex > prevIndex + 1) { + const dots = headers.map(() => "...").join("\t"); + formattedRows.push(`...\t${dots}`); + } + prevIndex = rowIndex; + + const record = tupleToRecord(tuple); + const cells = headers.map(h => { + const val = record[h]; + if (val === null) return "NaN"; + if (val === undefined) return ""; + if (typeof val === "number" || typeof val === "boolean") return String(val); + if (typeof val === "string") { + if (val === "NULL") return "NaN"; + return val.replace(/\t/g, "\\t").replace(/\n/g, "\\n"); + } + return JSON.stringify(val); + }); + formattedRows.push(`${rowIndex}\t${cells.join("\t")}`); + } + + return [headerLine, ...formattedRows].join("\n"); +} + +// Warnings are the console messages the engine tags with a "WARNING: " title +// prefix; derive them rather than carrying a separate field on the summary. +export function getOperatorWarnings(opInfo: OperatorExecutionSummary): string[] { + return (opInfo.consoleMessages ?? []).filter(m => m.title.startsWith("WARNING: ")).map(m => m.title); } export function createToolResult(message: string): string { diff --git a/agent-service/src/agent/tools/workflow-execution-tools.spec.ts b/agent-service/src/agent/tools/workflow-execution-tools.spec.ts new file mode 100644 index 00000000000..ff42e7f0144 --- /dev/null +++ b/agent-service/src/agent/tools/workflow-execution-tools.spec.ts @@ -0,0 +1,468 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from "bun:test"; +import { executeOperatorAndFormat, createExecuteOperatorTool, type ExecutionConfig } from "./workflow-execution-tools"; +import { WorkflowState } from "../workflow-state"; +import { WorkflowSystemMetadata } from "../util/workflow-system-metadata"; +import { + ConsoleMessageType, + OperatorResultMode, + OperatorState, + WorkflowExecutionState, + WorkflowFatalErrorType, + type OperatorExecutionSummary, + type Tuple, + type WorkflowExecutionSummary, + type WorkflowFatalError, +} from "../../types/execution"; +import type { OperatorLink, OperatorPredicate, PortDescription } from "../../types/workflow"; + +// --- fixtures ------------------------------------------------------------- + +// Build an engine-style Tuple (all-STRING schema) from a column->value record. +function recordToTuple(row: Record): Tuple { + return { + schema: { attributes: Object.keys(row).map(name => ({ attributeName: name, attributeType: "string" })) }, + fields: Object.values(row), + }; +} + +function makeOperator( + id: string, + opts: { inputPorts?: PortDescription[]; outputPorts?: PortDescription[]; operatorType?: string } = {} +): OperatorPredicate { + return { + operatorID: id, + operatorType: opts.operatorType ?? "TestOp", + operatorVersion: "1", + operatorProperties: {}, + inputPorts: opts.inputPorts ?? [], + outputPorts: opts.outputPorts ?? [{ portID: "out-0" }], + showAdvanced: false, + }; +} + +// A valid two-operator DAG: src (no inputs) --> tgt (one required input). +function makeLinearState(): { state: WorkflowState; source: string; target: string } { + const state = new WorkflowState(); + const source = "src"; + const target = "tgt"; + state.addOperator(makeOperator(source, { inputPorts: [], outputPorts: [{ portID: "out-0" }] })); + state.addOperator( + makeOperator(target, { + inputPorts: [{ portID: "in-0", disallowMultiInputs: true }], + outputPorts: [{ portID: "out-0" }], + }) + ); + const link: OperatorLink = { + linkID: "link-1", + source: { operatorID: source, portID: "out-0" }, + target: { operatorID: target, portID: "in-0" }, + }; + state.addLink(link); + return { state, source, target }; +} + +function makeConfig(overrides: Partial = {}): ExecutionConfig { + return { userToken: "tok", workflowId: 1, ...overrides }; +} + +function makeFatal(message: string): WorkflowFatalError { + return { + type: { name: WorkflowFatalErrorType.EXECUTION_FAILURE }, + timestamp: { seconds: 0, nanos: 0 }, + message, + details: "", + operatorId: "", + workerId: "", + }; +} + +// --- fetch / metadata stubbing ------------------------------------------- + +const originalFetch = globalThis.fetch; +let validateSpy: ReturnType; + +function setFetchResolving(value: Response): void { + globalThis.fetch = mock(async () => value) as unknown as typeof fetch; +} + +function setFetchRejecting(error: Error): void { + globalThis.fetch = mock(async () => { + throw error; + }) as unknown as typeof fetch; +} + +function jsonResponse(summary: WorkflowExecutionSummary): Response { + return new Response(JSON.stringify(summary), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); +} + +beforeEach(() => { + // Default: every operator's schema validates, so validation never short-circuits. + validateSpy = spyOn(WorkflowSystemMetadata.getInstance(), "validateOperatorProperties").mockReturnValue({ + isValid: true, + }); +}); + +afterEach(() => { + globalThis.fetch = originalFetch; + validateSpy.mockRestore(); +}); + +// --- tests ---------------------------------------------------------------- + +describe("executeOperatorAndFormat - pre-flight guards", () => { + test("returns an error when the target operator is absent from the workflow", async () => { + const state = new WorkflowState(); + const out = await executeOperatorAndFormat(state, makeConfig(), "missing"); + expect(out).toBe("[ERROR] Cannot execute: workflow has no operators."); + }); + + test("blocks on the target operator's schema validation errors", async () => { + const { state, target } = makeLinearState(); + validateSpy.mockReturnValue({ isValid: false, messages: { attributeName: "must not be empty" } }); + + const out = await executeOperatorAndFormat(state, makeConfig(), target); + expect(out).toContain("[ERROR]"); + expect(out).toContain(`Operator ${target}:`); + expect(out).toContain("- attributeName: must not be empty"); + }); + + test("blocks when a disallow-multi-input port has no incoming link", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("solo", { inputPorts: [{ portID: "in-0", disallowMultiInputs: true }] })); + + const out = await executeOperatorAndFormat(state, makeConfig(), "solo"); + expect(out).toContain("[ERROR]"); + expect(out).toContain("requires 1 input, has 0"); + }); + + test("blocks when a regular input port has no incoming link", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("solo", { inputPorts: [{ portID: "in-0" }] })); + + const out = await executeOperatorAndFormat(state, makeConfig(), "solo"); + expect(out).toContain("[ERROR]"); + expect(out).toContain("requires at least 1 input, has 0"); + }); +}); + +describe("executeOperatorAndFormat - successful runs", () => { + test("renders the shape, warnings, gaps, and every cell type", async () => { + const { state, source, target } = makeLinearState(); + + const sampleTuples: [number, Tuple][] = [ + [ + 0, + recordToTuple({ + num: 1, + bool: true, + str: "hello", + nil: null, + nullstr: "NULL", + escaped: "a\tb\nc", + obj: { k: 1 }, + }), + ], + [ + 3, // gap after 0 -> a "..." separator row is inserted + recordToTuple({ + num: 2, + bool: false, + str: "world", + nil: null, + nullstr: "kept", + escaped: "plain", + obj: [1, 2], + }), + ], + ]; + + const summary: WorkflowExecutionSummary = { + success: true, + state: WorkflowExecutionState.COMPLETED, + operators: { + [target]: { + state: OperatorState.COMPLETED, + errorMessages: [], + resultSummary: { resultMode: OperatorResultMode.TABLE, sampleTuples, totalTuplesCount: 10 }, + consoleMessages: [ + { msgType: ConsoleMessageType.PRINT, title: "WARNING: truncated output", message: "" }, + { msgType: ConsoleMessageType.PRINT, title: "just info, not a warning", message: "" }, + ], + }, + [source]: { state: OperatorState.COMPLETED, errorMessages: [] }, + // An errored sibling: the notify loop must skip it (covers the false branch). + ghost: { state: OperatorState.FAILED, errorMessages: [makeFatal("ignored")] }, + }, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const onResult = mock(() => {}); + const out = await executeOperatorAndFormat(state, makeConfig(), target, { onResult }); + + expect(out).toContain(`Executed operator ${target}`); + expect(out).toContain("Output table shape: (10, 7)"); + expect(out).toContain("WARNING: truncated output"); + expect(out).not.toContain("just info"); + expect(out).toContain("NaN"); // null cell + expect(out).toContain("a\\tb\\nc"); // tab/newline escaped + expect(out).toContain('{"k":1}'); // object serialized + expect(out).toContain("...\t"); // gap separator + + // Notified for the two clean operators only. + const notified = onResult.mock.calls.map(c => (c as unknown as [string])[0]); + expect(notified.sort()).toEqual([source, target]); + }); + + test("returns a placeholder when the operator produced no result summary", async () => { + const { state, target } = makeLinearState(); + const summary: WorkflowExecutionSummary = { + success: true, + state: WorkflowExecutionState.COMPLETED, + operators: { [target]: { state: OperatorState.COMPLETED, errorMessages: [] } }, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const out = await executeOperatorAndFormat(state, makeConfig(), target); + expect(out).toBe("(no result data)"); + }); + + test("emits only the shape line when the result has zero sample rows", async () => { + const { state, target } = makeLinearState(); + const summary: WorkflowExecutionSummary = { + success: true, + state: WorkflowExecutionState.COMPLETED, + operators: { + [target]: { + state: OperatorState.COMPLETED, + errorMessages: [], + resultSummary: { resultMode: OperatorResultMode.TABLE, sampleTuples: [], totalTuplesCount: 0 }, + }, + }, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const out = await executeOperatorAndFormat(state, makeConfig(), target); + expect(out).toContain(`Executed operator ${target}`); + expect(out).toContain("Output table shape: (0, 0)"); + }); + + test("truncates output that exceeds the character budget while keeping the header", async () => { + const { state, target } = makeLinearState(); + const sampleTuples: [number, Tuple][] = Array.from({ length: 12 }, (_, i) => [ + i, + recordToTuple({ col: `value-${i}` }), + ]); + const summary: WorkflowExecutionSummary = { + success: true, + state: WorkflowExecutionState.COMPLETED, + operators: { + [target]: { + state: OperatorState.COMPLETED, + errorMessages: [], + resultSummary: { resultMode: OperatorResultMode.TABLE, sampleTuples, totalTuplesCount: 12 }, + }, + }, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const out = await executeOperatorAndFormat(state, makeConfig({ maxOperatorResultCharLimit: 40 }), target); + expect(out).toContain("col"); // header retained + // Not every one of the 12 rows survives the budget. + const dataRowCount = out.split("\n").filter(l => /^\d+\t/.test(l)).length; + expect(dataRowCount).toBeLessThan(12); + }); +}); + +describe("executeOperatorAndFormat - execution failures", () => { + test("formats per-operator errors and notifies with a synthetic failure summary (FAILED)", async () => { + const { state, target } = makeLinearState(); + const summary: WorkflowExecutionSummary = { + success: false, + state: WorkflowExecutionState.FAILED, + operators: { + [target]: { state: OperatorState.FAILED, errorMessages: [makeFatal("kaboom")] }, + }, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const onResult = mock(() => {}); + const out = await executeOperatorAndFormat(state, makeConfig(), target, { onResult }); + + expect(out).toContain("[ERROR]"); + expect(out).toContain("Execution error:"); + expect(out).toContain("kaboom"); + + expect(onResult).toHaveBeenCalledTimes(1); + const [, info] = onResult.mock.calls[0] as unknown as [string, OperatorExecutionSummary]; + expect(info.state).toBe(OperatorState.FAILED); + expect(info.errorMessages[0].type.name).toBe(WorkflowFatalErrorType.EXECUTION_FAILURE); + expect(info.errorMessages[0].message).toContain("kaboom"); + }); + + test("reports a timeout when the workflow was KILLED (no onResult callback)", async () => { + const { state, target } = makeLinearState(); + const summary: WorkflowExecutionSummary = { + success: false, + state: WorkflowExecutionState.KILLED, + operators: {}, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const out = await executeOperatorAndFormat(state, makeConfig(), target); + expect(out).toContain("[ERROR]"); + expect(out).toContain("Workflow execution was killed (timeout)."); + }); + + test("surfaces general errors when the HTTP request itself fails (ERROR state)", async () => { + const { state, target } = makeLinearState(); + setFetchResolving(new Response("backend detail", { status: 500, statusText: "Server Error" })); + + const out = await executeOperatorAndFormat(state, makeConfig(), target); + expect(out).toContain("[ERROR]"); + expect(out).toContain("Execution request failed: 500"); + expect(out).toContain("backend detail"); + }); + + test("surfaces the message when fetch rejects with a non-abort error", async () => { + const { state, target } = makeLinearState(); + setFetchRejecting(new Error("network down")); + + const out = await executeOperatorAndFormat(state, makeConfig(), target); + expect(out).toContain("[ERROR]"); + expect(out).toContain("network down"); + }); + + test("returns a 'no result' error when the operator is missing from a successful run", async () => { + const { state, target } = makeLinearState(); + const summary: WorkflowExecutionSummary = { + success: true, + state: WorkflowExecutionState.COMPLETED, + operators: {}, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const out = await executeOperatorAndFormat(state, makeConfig(), target); + expect(out).toContain("[ERROR]"); + expect(out).toContain(`No result found for operator: ${target}`); + }); + + test("joins operator error messages when a successful run still reports operator errors", async () => { + const { state, target } = makeLinearState(); + const summary: WorkflowExecutionSummary = { + success: true, + state: WorkflowExecutionState.COMPLETED, + operators: { + [target]: { state: OperatorState.FAILED, errorMessages: [makeFatal("e1"), makeFatal("e2")] }, + }, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const onResult = mock(() => {}); + const out = await executeOperatorAndFormat(state, makeConfig(), target, { onResult }); + expect(out).toContain("[ERROR]"); + expect(out).toContain("e1; e2"); + expect(onResult).toHaveBeenCalledTimes(1); + }); +}); + +describe("executeOperatorAndFormat - abort and callback failures", () => { + test("re-throws AbortError raised by the HTTP layer", async () => { + const { state, target } = makeLinearState(); + const abortErr = new Error("aborted"); + abortErr.name = "AbortError"; + setFetchRejecting(abortErr); + + await expect(executeOperatorAndFormat(state, makeConfig(), target)).rejects.toThrow("aborted"); + }); + + test("wraps a throwing onResult callback into an error result", async () => { + const { state, target } = makeLinearState(); + const summary: WorkflowExecutionSummary = { + success: true, + state: WorkflowExecutionState.COMPLETED, + operators: { + [target]: { + state: OperatorState.COMPLETED, + errorMessages: [], + resultSummary: { + resultMode: OperatorResultMode.TABLE, + sampleTuples: [[0, recordToTuple({ col: "v" })]], + totalTuplesCount: 1, + }, + }, + }, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const onResult = () => { + throw new Error("cb boom"); + }; + const out = await executeOperatorAndFormat(state, makeConfig(), target, { onResult }); + expect(out).toContain("[ERROR] Execution failed: cb boom"); + }); +}); + +describe("createExecuteOperatorTool", () => { + test("builds a tool whose execute delegates to executeOperatorAndFormat", async () => { + const { state, target } = makeLinearState(); + const summary: WorkflowExecutionSummary = { + success: true, + state: WorkflowExecutionState.COMPLETED, + operators: { + [target]: { + state: OperatorState.COMPLETED, + errorMessages: [], + resultSummary: { + resultMode: OperatorResultMode.TABLE, + sampleTuples: [[0, recordToTuple({ col: "v" })]], + totalTuplesCount: 1, + }, + }, + }, + errors: [], + }; + setFetchResolving(jsonResponse(summary)); + + const onResult = mock(() => {}); + const executeTool = createExecuteOperatorTool(state, () => makeConfig({ computingUnitId: 2 }), onResult); + expect(typeof executeTool.execute).toBe("function"); + + const out = await (executeTool.execute as (a: { operatorId: string }, o: object) => Promise)( + { operatorId: target }, + {} + ); + expect(out).toContain(`Executed operator ${target}`); + expect(onResult).toHaveBeenCalledWith(target, expect.anything()); + }); +}); diff --git a/agent-service/src/agent/tools/workflow-execution-tools.ts b/agent-service/src/agent/tools/workflow-execution-tools.ts index 78c6cfa3d55..f564e53bcb8 100644 --- a/agent-service/src/agent/tools/workflow-execution-tools.ts +++ b/agent-service/src/agent/tools/workflow-execution-tools.ts @@ -19,12 +19,26 @@ import { z } from "zod"; import { tool } from "ai"; -import { createErrorResult, formatExecuteOperatorResult, getVisibleResultHeaders } from "./tools-utility"; +import { + createErrorResult, + formatExecuteOperatorResult, + formatSampleRowsAsTsv, + getOperatorErrorText, + getOperatorWarnings, + tupleColumns, +} from "./tools-utility"; import type { WorkflowState } from "../workflow-state"; import { getBackendConfig } from "../../api/backend-api"; import { env } from "../../config/env"; import type { LogicalPlan, LogicalLink } from "../../api/execution-api"; -import type { OperatorInfo, SyncExecutionResult } from "../../types/execution"; +import { + OperatorState, + WorkflowFatalErrorType, + WorkflowExecutionState, + type OperatorExecutionSummary, + type WorkflowFatalError, + type WorkflowExecutionSummary, +} from "../../types/execution"; import { WorkflowSystemMetadata } from "../util/workflow-system-metadata"; import { DEFAULT_AGENT_SETTINGS } from "../../types/agent"; import { createLogger } from "../../logger"; @@ -255,7 +269,7 @@ async function executeWorkflowHttp( config: ExecutionConfig, logicalPlan: LogicalPlan, options: { abortSignal?: AbortSignal } = {} -): Promise { +): Promise { const backendConfig = getBackendConfig(); const workflowId = config.workflowId; @@ -312,7 +326,7 @@ async function executeWorkflowHttp( throw new Error(`Execution request failed: ${response.status} ${response.statusText} - ${errorText}`); } - return (await response.json()) as SyncExecutionResult; + return (await response.json()) as WorkflowExecutionSummary; } catch (error) { if (error instanceof Error && error.name === "AbortError") { throw error; @@ -320,62 +334,19 @@ async function executeWorkflowHttp( log.error({ err: error }, "execution failed"); return { success: false, - state: "Error", + state: WorkflowExecutionState.ERROR, operators: {}, - errors: [error instanceof Error ? error.message : "Unknown error"], + errors: [makeExecutionFailure(error instanceof Error ? error.message : "Unknown error", "")], }; } } -function formatInputOutput( - workflowState: WorkflowState, - operatorId: string, - opInfo: OperatorInfo, - outputColumns: number -): string { - const outputRows = opInfo.totalRowCount ?? opInfo.outputTuples; - const outputLine = `Output table shape: (${outputRows}, ${outputColumns})`; - - const inputShapes = opInfo.inputPortShapes; - if (!inputShapes || inputShapes.length === 0) { - return outputLine; - } - - const inputLinks = workflowState.getAllLinks().filter(l => l.target.operatorID === operatorId); - const portIndexToUpstream = new Map(); - const op = workflowState.getOperator(operatorId); - for (const link of inputLinks) { - const portIdx = op?.inputPorts.findIndex(p => p.portID === link.target.portID) ?? -1; - if (portIdx >= 0) { - portIndexToUpstream.set(portIdx, link.source.operatorID); - } - } - - const inputPart = inputShapes - .sort((a, b) => a.portIndex - b.portIndex) - .map(p => { - const name = portIndexToUpstream.get(p.portIndex) ?? `input${p.portIndex}`; - return `${name}(${p.rows}, ${p.columns})`; - }) - .join(", "); - - return `Input operator(table shape): ${inputPart}\n${outputLine}`; -} - function formatExecutionError( - compilationErrors?: Record, operatorErrors?: Array<{ operatorId: string; error: string }>, - generalErrors?: string[] + generalErrors?: WorkflowFatalError[] ): string { const lines: string[] = ["Execution failed due to the following error:"]; - if (compilationErrors && Object.keys(compilationErrors).length > 0) { - lines.push("Compilation error:"); - for (const [key, value] of Object.entries(compilationErrors)) { - lines.push(` ${key}: ${value}`); - } - } - if (operatorErrors && operatorErrors.length > 0) { lines.push("Execution error:"); for (const { operatorId, error } of operatorErrors) { @@ -386,50 +357,23 @@ function formatExecutionError( if (generalErrors && generalErrors.length > 0) { lines.push("Error:"); for (const error of generalErrors) { - lines.push(` ${error}`); + lines.push(` ${error.message}`); } } return lines.join("\n"); } -function jsonToTableFormat(jsonResult: Record[]): string { - if (!jsonResult || jsonResult.length === 0) return ""; - - const hasRowIndex = jsonResult.length > 0 && "__row_index__" in jsonResult[0]; - const headers = getVisibleResultHeaders(jsonResult[0]); - if (headers.length === 0) return ""; - // Leading tab aligns headers with the index column (pandas __repr__ style). - const headerLine = "\t" + headers.join("\t"); - - const formattedRows: string[] = []; - let prevIndex = -1; - - for (let i = 0; i < jsonResult.length; i++) { - const row = jsonResult[i]; - const rowIndex = hasRowIndex ? (row["__row_index__"] as number) : i; - - if (prevIndex >= 0 && rowIndex > prevIndex + 1) { - const dots = headers.map(() => "...").join("\t"); - formattedRows.push(`...\t${dots}`); - } - prevIndex = rowIndex; - - const cells = headers.map(h => { - const val = row[h]; - if (val === null) return "NaN"; - if (val === undefined) return ""; - if (typeof val === "number" || typeof val === "boolean") return String(val); - if (typeof val === "string") { - if (val === "NULL") return "NaN"; - return val.replace(/\t/g, "\\t").replace(/\n/g, "\\n"); - } - return JSON.stringify(val); - }); - formattedRows.push(`${rowIndex}\t${cells.join("\t")}`); - } - - return [headerLine, ...formattedRows].join("\n"); +function makeExecutionFailure(message: string, operatorId: string): WorkflowFatalError { + const now = Date.now(); + return { + type: { name: WorkflowFatalErrorType.EXECUTION_FAILURE }, + timestamp: { seconds: Math.floor(now / 1000), nanos: (now % 1000) * 1_000_000 }, + message, + details: "", + operatorId, + workerId: "", + }; } export async function executeOperatorAndFormat( @@ -438,7 +382,7 @@ export async function executeOperatorAndFormat( operatorId: string, options: { abortSignal?: AbortSignal; - onResult?: (operatorId: string, operatorInfo: OperatorInfo) => void; + onResult?: (operatorId: string, operatorInfo: OperatorExecutionSummary) => void; onResultLegacy?: (operatorId: string, backendStats?: Record) => void; } = {} ): Promise { @@ -466,34 +410,29 @@ export async function executeOperatorAndFormat( } } - const result: SyncExecutionResult = await executeWorkflowHttp(config, logicalPlan, { + const result: WorkflowExecutionSummary = await executeWorkflowHttp(config, logicalPlan, { abortSignal: options.abortSignal, }); if (!result.success) { - const compilationErrors = - result.state === "CompilationFailed" || result.state === "ValidationFailed" - ? result.compilationErrors - : undefined; - const operatorErrors = - result.state === "Failed" + result.state === WorkflowExecutionState.FAILED ? Object.entries(result.operators) - .filter(([_, op]) => op.error) - .map(([opId, op]) => ({ operatorId: opId, error: op.error! })) + .map(([opId, op]) => ({ operatorId: opId, error: getOperatorErrorText(op) })) + .filter(({ error }) => error) : undefined; - const generalErrors = result.state === "Killed" ? ["Workflow execution was killed (timeout)."] : result.errors; + const generalErrors = + result.state === WorkflowExecutionState.KILLED + ? [makeExecutionFailure("Workflow execution was killed (timeout).", "")] + : result.errors; - const errorText = formatExecutionError(compilationErrors, operatorErrors, generalErrors); + const errorText = formatExecutionError(operatorErrors, generalErrors); if (options.onResult) { - const errorInfo: OperatorInfo = { - state: result.state, - inputTuples: 0, - outputTuples: 0, - resultMode: "table", - error: errorText, + const errorInfo: OperatorExecutionSummary = { + state: OperatorState.FAILED, + errorMessages: [makeExecutionFailure(errorText, operatorId)], }; options.onResult(operatorId, errorInfo); } @@ -504,35 +443,36 @@ export async function executeOperatorAndFormat( const opInfo = result.operators[operatorId]; if (!opInfo) { return createErrorResult( - formatExecutionError(undefined, undefined, [`No result found for operator: ${operatorId}`]) + formatExecutionError(undefined, [makeExecutionFailure(`No result found for operator: ${operatorId}`, "")]) ); } - if (opInfo.error) { + const opError = getOperatorErrorText(opInfo); + if (opError) { if (options.onResult) { options.onResult(operatorId, opInfo); } - return createErrorResult(formatExecutionError(undefined, [{ operatorId, error: opInfo.error }])); + return createErrorResult(formatExecutionError([{ operatorId, error: opError }])); } - if (!opInfo.result || !Array.isArray(opInfo.result)) { + const sampleTuples = opInfo.resultSummary?.sampleTuples; + if (!sampleTuples || !Array.isArray(sampleTuples)) { return "(no result data)"; } - const jsonArray = opInfo.result as Record[]; - const headers = jsonArray.length > 0 ? getVisibleResultHeaders(jsonArray[0]) : []; + const headers = sampleTuples.length > 0 ? tupleColumns(sampleTuples[0][1]) : []; const columns = headers.length; // Notify for every operator in the execution so upstream stats are also stored. if (options.onResult) { for (const [opId, info] of Object.entries(result.operators)) { - if (info && !info.error) { + if (info && !getOperatorErrorText(info)) { options.onResult(opId, info); } } } - let dataString = jsonToTableFormat(jsonArray); + let dataString = formatSampleRowsAsTsv(sampleTuples); // Safety-net: TSV serialization may add padding beyond backend's raw-record budget. const charLimit = config.maxOperatorResultCharLimit ?? DEFAULT_AGENT_SETTINGS.maxOperatorResultCharLimit; @@ -568,9 +508,11 @@ export async function executeOperatorAndFormat( dataString = [headerLine, ...keptRows].join("\n"); } - const shapeLine = formatInputOutput(workflowState, operatorId, opInfo, columns); + // Output shape only; the agent derives input-port shapes from the DAG + the + // upstream operators' own output shapes shown in context. + const shapeLine = `Output table shape: (${opInfo.resultSummary?.totalTuplesCount ?? 0}, ${columns})`; - const warningLines = opInfo.warnings?.map(w => w) ?? []; + const warningLines = getOperatorWarnings(opInfo); const metadataLines = [shapeLine, ...warningLines].filter(Boolean); @@ -589,7 +531,7 @@ export async function executeOperatorAndFormat( export function createExecuteOperatorTool( workflowState: WorkflowState, getConfig: () => ExecutionConfig, - onResult?: (operatorId: string, operatorInfo: OperatorInfo) => void + onResult?: (operatorId: string, operatorInfo: OperatorExecutionSummary) => void ) { return tool({ description: diff --git a/agent-service/src/agent/workflow-result-state.spec.ts b/agent-service/src/agent/workflow-result-state.spec.ts index b2e46fd0d9e..7068ddf671a 100644 --- a/agent-service/src/agent/workflow-result-state.spec.ts +++ b/agent-service/src/agent/workflow-result-state.spec.ts @@ -19,14 +19,13 @@ import { describe, expect, test } from "bun:test"; import { WorkflowResultState } from "./workflow-result-state"; -import type { OperatorInfo } from "../types/execution"; +import { OperatorResultMode, OperatorState, type OperatorExecutionSummary } from "../types/execution"; -function makeInfo(outputTuples: number): OperatorInfo { +function makeInfo(totalTuplesCount: number): OperatorExecutionSummary { return { - state: "Completed", - inputTuples: 0, - outputTuples, - resultMode: "table", + state: OperatorState.COMPLETED, + errorMessages: [], + resultSummary: { resultMode: OperatorResultMode.TABLE, sampleTuples: [], totalTuplesCount }, }; } @@ -40,15 +39,15 @@ describe("WorkflowResultState - ancestor walk", () => { state.set("op1", "step-C", makeInfo(3)); path = ["step-A", "step-B", "step-C"]; - expect(state.get("op1")?.operatorInfo.outputTuples).toBe(3); + expect(state.get("op1")?.operatorInfo.resultSummary?.totalTuplesCount).toBe(3); // Rewind to step-B; step-C is no longer an ancestor. path = ["step-A", "step-B"]; - expect(state.get("op1")?.operatorInfo.outputTuples).toBe(2); + expect(state.get("op1")?.operatorInfo.resultSummary?.totalTuplesCount).toBe(2); // Rewind further. path = ["step-A"]; - expect(state.get("op1")?.operatorInfo.outputTuples).toBe(1); + expect(state.get("op1")?.operatorInfo.resultSummary?.totalTuplesCount).toBe(1); }); test("returns undefined when no ancestor has a result", () => { @@ -74,8 +73,8 @@ describe("WorkflowResultState - ancestor walk", () => { path = ["step-A", "step-B"]; const visible = state.getAllVisible(); expect(visible.size).toBe(2); - expect(visible.get("op1")?.operatorInfo.outputTuples).toBe(1); - expect(visible.get("op2")?.operatorInfo.outputTuples).toBe(7); + expect(visible.get("op1")?.operatorInfo.resultSummary?.totalTuplesCount).toBe(1); + expect(visible.get("op2")?.operatorInfo.resultSummary?.totalTuplesCount).toBe(7); }); test("clear drops all stored results", () => { @@ -90,6 +89,18 @@ describe("WorkflowResultState - ancestor walk", () => { const state = new WorkflowResultState(() => ["step-A"]); state.set("op1", "step-A", makeInfo(1)); state.set("op1", "step-A", makeInfo(42)); - expect(state.get("op1")?.operatorInfo.outputTuples).toBe(42); + expect(state.get("op1")?.operatorInfo.resultSummary?.totalTuplesCount).toBe(42); + }); + + test("getOperatorInfo returns the visible operator summary", () => { + const state = new WorkflowResultState(() => ["step-A"]); + state.set("op1", "step-A", makeInfo(7)); + expect(state.getOperatorInfo("op1")?.resultSummary?.totalTuplesCount).toBe(7); + }); + + test("getOperatorInfo returns undefined when nothing is visible", () => { + const state = new WorkflowResultState(() => ["step-X"]); + state.set("op1", "step-A", makeInfo(1)); + expect(state.getOperatorInfo("op1")).toBeUndefined(); }); }); diff --git a/agent-service/src/agent/workflow-result-state.ts b/agent-service/src/agent/workflow-result-state.ts index e6f13c2301a..34d604657e3 100644 --- a/agent-service/src/agent/workflow-result-state.ts +++ b/agent-service/src/agent/workflow-result-state.ts @@ -17,10 +17,10 @@ * under the License. */ -import type { OperatorInfo } from "../types/execution"; +import type { OperatorExecutionSummary } from "../types/execution"; interface ResultEntry { - operatorInfo: OperatorInfo; + operatorInfo: OperatorExecutionSummary; stepId: string; } @@ -37,7 +37,7 @@ export class WorkflowResultState { constructor(private getAncestorPath: () => string[]) {} - set(operatorId: string, stepId: string, operatorInfo: OperatorInfo): void { + set(operatorId: string, stepId: string, operatorInfo: OperatorExecutionSummary): void { let versions = this.results.get(operatorId); if (!versions) { versions = new Map(); @@ -58,7 +58,7 @@ export class WorkflowResultState { return undefined; } - getOperatorInfo(operatorId: string): OperatorInfo | undefined { + getOperatorInfo(operatorId: string): OperatorExecutionSummary | undefined { return this.get(operatorId)?.operatorInfo; } diff --git a/agent-service/src/api/compile-api.ts b/agent-service/src/api/compile-api.ts index 8ffd27fd52c..8a361231c42 100644 --- a/agent-service/src/api/compile-api.ts +++ b/agent-service/src/api/compile-api.ts @@ -19,8 +19,13 @@ import { getBackendConfig } from "./backend-api"; import type { LogicalPlan, OperatorPortSchemaMap } from "../types/workflow"; +import type { WorkflowFatalError } from "../types/execution"; import { createLogger } from "../logger"; +// WorkflowFatalError is defined in types/execution.ts (shared by compile and +// execution errors); re-exported here for existing importers of this module. +export type { WorkflowFatalError }; + const log = createLogger("CompileAPI"); export interface SchemaAttribute { @@ -30,12 +35,6 @@ export interface SchemaAttribute { export type PortSchema = ReadonlyArray; -export interface WorkflowFatalError { - type: string; - message: string; - operatorId?: string; -} - export interface WorkflowCompilationResponse { physicalPlan?: any; operatorOutputSchemas: Record; diff --git a/agent-service/src/server.spec.ts b/agent-service/src/server.spec.ts index c1fdddc9339..ff3e65e6fcc 100644 --- a/agent-service/src/server.spec.ts +++ b/agent-service/src/server.spec.ts @@ -21,6 +21,7 @@ import { beforeEach, describe, expect, spyOn, test } from "bun:test"; import { buildApp, start, _resetAgentStoreForTests, _getAgentForTests } from "./server"; import { WorkflowSystemMetadata } from "./agent/util/workflow-system-metadata"; import { env } from "./config/env"; +import { OperatorResultMode, OperatorState } from "./types/execution"; const API = env.API_PREFIX; const app = buildApp(); @@ -323,6 +324,8 @@ describe("agent read routes", () => { test("GET /:id/operator-results maps the visible operator results", async () => { const agent = _getAgentForTests(id)!; + // PR1 keeps the frontend-facing route on the legacy summary shape while + // agent-service stores the new OperatorExecutionSummary internally. (agent as any).getWorkflowResultState = () => ({ getAllVisible: () => new Map([ @@ -330,27 +333,41 @@ describe("agent read routes", () => { "op-1", { operatorInfo: { - state: "COMPLETED", - inputTuples: 1, - outputTuples: 2, - inputPortShapes: [], - result: [{ a: 1 }], - error: undefined, - warnings: [], - consoleLogs: [], - totalRowCount: 2, - resultStatistics: {}, + state: OperatorState.COMPLETED, + errorMessages: [], + resultSummary: { + resultMode: OperatorResultMode.TABLE, + sampleTuples: [ + [0, { schema: { attributes: [{ attributeName: "a", attributeType: "string" }] }, fields: [1] }], + ], + totalTuplesCount: 2, + }, + consoleMessages: [{ msgType: "PRINT", title: "WARNING: check me", message: "" }], }, }, ], ]), }); - const body = await readJson<{ results: Record }>( - await getJson(`${API}/agents/${id}/operator-results`) - ); + const body = await readJson<{ + results: Record< + string, + { + state: string; + outputTuples: number; + outputColumns: number; + totalRowCount: number; + warnings: string[]; + sampleRecords: Record[]; + } + >; + }>(await getJson(`${API}/agents/${id}/operator-results`)); + expect(body.results["op-1"].state).toBe("Completed"); expect(body.results["op-1"].outputTuples).toBe(2); expect(body.results["op-1"].outputColumns).toBe(1); + expect(body.results["op-1"].totalRowCount).toBe(2); + expect(body.results["op-1"].warnings).toEqual(["WARNING: check me"]); + expect(body.results["op-1"].sampleRecords).toEqual([{ __row_index__: 0, a: 1 }]); }); }); diff --git a/agent-service/src/server.ts b/agent-service/src/server.ts index 030d27b95bf..077a3631f74 100644 --- a/agent-service/src/server.ts +++ b/agent-service/src/server.ts @@ -21,7 +21,6 @@ import { Elysia, t } from "elysia"; import { cors } from "@elysiajs/cors"; import { createOpenAI } from "@ai-sdk/openai"; import { TexeraAgent } from "./agent/texera-agent"; -import { getVisibleResultHeaders } from "./agent/tools/tools-utility"; import { getBackendConfig } from "./api/backend-api"; import { extractBearerToken, extractUserFromToken, validateToken } from "./api/auth-api"; import { retrieveWorkflow } from "./api/workflow-api"; @@ -42,7 +41,7 @@ import type { import { AgentState, OperatorResultSerializationMode } from "./types/agent"; import type { WsClientCommand, WsServerEvent } from "./types/ws"; import { WsServerSnapshotEvent, WsServerStepEvent, WsServerStatusEvent, WsServerErrorEvent } from "./types/ws"; -import type { OperatorResultSummary } from "./types/execution"; +import { OperatorResultMode, type OperatorExecutionSummary, type Tuple } from "./types/execution"; const agentStore = new Map(); let agentCounter = 0; @@ -388,25 +387,66 @@ const agentsRouter = new Elysia({ prefix: "/agents" }) } ); -function getOperatorResultSummaries(agent: TexeraAgent): Record { +interface LegacyOperatorResultSummary { + state: string; + inputTuples: number; + outputTuples: number; + inputPortShapes?: { portIndex: number; rows: number; columns: number }[]; + outputColumns?: number; + error?: string; + warnings?: string[]; + consoleLogCount?: number; + totalRowCount?: number; + sampleRecords?: Record[]; + resultStatistics?: Record; +} + +function tupleToRecord(tuple: Tuple): Record { + const record: Record = {}; + tuple.schema.attributes.forEach((a, i) => { + record[a.attributeName] = tuple.fields[i]; + }); + return record; +} + +function toLegacyOperatorResultSummary(opInfo: OperatorExecutionSummary): LegacyOperatorResultSummary { + const error = opInfo.errorMessages + .map(e => e.message) + .filter(Boolean) + .join("; "); + const warnings = (opInfo.consoleMessages ?? []).filter(m => m.title.startsWith("WARNING: ")).map(m => m.title); + const sampleRecords: Record[] | undefined = opInfo.resultSummary?.sampleTuples.map(([rowIndex, tuple]) => ({ + __row_index__: rowIndex, + ...tupleToRecord(tuple), + })); + + if ( + sampleRecords && + sampleRecords.length > 0 && + opInfo.resultSummary?.resultMode === OperatorResultMode.VISUALIZATION + ) { + sampleRecords[0]["__is_visualization__"] = true; + } + + return { + state: opInfo.state, + inputTuples: 0, + outputTuples: opInfo.resultSummary?.totalTuplesCount ?? 0, + outputColumns: opInfo.resultSummary?.sampleTuples[0]?.[1].schema.attributes.length, + error: error || undefined, + warnings: warnings.length > 0 ? warnings : undefined, + consoleLogCount: opInfo.consoleMessages?.length, + totalRowCount: opInfo.resultSummary?.totalTuplesCount, + sampleRecords, + }; +} + +function getOperatorResultSummaries(agent: TexeraAgent): Record { const resultState = agent.getWorkflowResultState(); const visible = resultState.getAllVisible(); - const results: Record = {}; + const results: Record = {}; for (const [opId, entry] of visible) { - const info = entry.operatorInfo; - results[opId] = { - state: info.state, - inputTuples: info.inputTuples, - outputTuples: info.outputTuples, - inputPortShapes: info.inputPortShapes, - outputColumns: info.result && info.result.length > 0 ? getVisibleResultHeaders(info.result[0]).length : undefined, - error: info.error, - warnings: info.warnings, - consoleLogCount: info.consoleLogs?.length, - totalRowCount: info.totalRowCount, - sampleRecords: info.result, - resultStatistics: info.resultStatistics, - }; + results[opId] = toLegacyOperatorResultSummary(entry.operatorInfo); } return results; } diff --git a/agent-service/src/types/execution.ts b/agent-service/src/types/execution.ts index d638a889d47..d7bf2915c4d 100644 --- a/agent-service/src/types/execution.ts +++ b/agent-service/src/types/execution.ts @@ -17,56 +17,124 @@ * under the License. */ -interface ConsoleMessage { - msgType: string; +export enum WorkflowFatalErrorType { + COMPILATION_ERROR = "COMPILATION_ERROR", + EXECUTION_FAILURE = "EXECUTION_FAILURE", +} + +// A fatal error reported for one operator. Reuses the engine's wire shape +// (workflowruntimestate.proto). The same type the workflow-compiling service +// returns for compilation errors, so compile and execution errors share one +// shape. Re-exported by api/compile-api.ts. +export interface WorkflowFatalError { + type: { name: WorkflowFatalErrorType }; + timestamp: { seconds: number; nanos: number }; message: string; + details: string; + operatorId: string; + workerId: string; } -interface PortShape { - portIndex: number; - rows: number; - columns: number; +// Lifecycle state of a single operator, as reported by the engine +// (mirrors the backend's WorkflowAggregatedState string mapping). +export enum OperatorState { + UNINITIALIZED = "Uninitialized", + READY = "Ready", + RUNNING = "Running", + PAUSING = "Pausing", + PAUSED = "Paused", + RESUMING = "Resuming", + COMPLETED = "Completed", + FAILED = "Failed", + KILLED = "Killed", + TERMINATED = "Terminated", + UNKNOWN = "Unknown", } -export interface OperatorInfo { - state: string; - inputTuples: number; - outputTuples: number; - inputPortShapes?: PortShape[]; - resultMode: string; - result?: Record[]; - totalRowCount?: number; - displayedRows?: number; - truncated?: boolean; - consoleLogs?: ConsoleMessage[]; - error?: string; - warnings?: string[]; - resultStatistics?: Record; +// Aggregated state of a whole workflow execution: the OperatorState values the +// engine reports, plus the synthetic outcomes the sync-execution endpoint adds. +export enum WorkflowExecutionState { + UNINITIALIZED = "Uninitialized", + READY = "Ready", + RUNNING = "Running", + PAUSING = "Pausing", + PAUSED = "Paused", + RESUMING = "Resuming", + COMPLETED = "Completed", + FAILED = "Failed", + KILLED = "Killed", + TERMINATED = "Terminated", + UNKNOWN = "Unknown", + ERROR = "Error", + COMPILATION_FAILED = "CompilationFailed", } -export interface SyncExecutionResult { - success: boolean; - state: string; - operators: Record; - compilationErrors?: Record; - errors?: string[]; +export enum ConsoleMessageType { + PRINT = "PRINT", + ERROR = "ERROR", + COMMAND = "COMMAND", + DEBUGGER = "DEBUGGER", } -/** - * Wire projection of one operator's execution result, summarized for the - * client: counts and a small record sample instead of full payloads. Returned - * by the REST route `GET /agents/:id/operator-results`. - */ +// A reduced console-message projection for sync-execution summaries. The engine +// proto also has workerId/timestamp/source; this summary keeps only the fields +// consumed by agent-service. +export interface ConsoleMessageSummary { + msgType: ConsoleMessageType; + title: string; + message: string; +} + +// A result row, mirroring the engine's Tuple wire shape +// (org.apache.texera.amber.core.tuple.Tuple): a schema plus positional field values. +// Because sampled rows are truncated (typed values become display strings), each Tuple +// carries a synthetic all-STRING schema over its columns. +export interface Attribute { + attributeName: string; + attributeType: string; +} + +export interface Schema { + attributes: Attribute[]; +} + +export interface Tuple { + schema: Schema; + fields: unknown[]; +} + +export enum OperatorResultMode { + TABLE = "table", + VISUALIZATION = "visualization", +} + +// An operator's output summary. Sample tuples carry their original row index. export interface OperatorResultSummary { - state: string; - inputTuples: number; - outputTuples: number; - inputPortShapes?: PortShape[]; - outputColumns?: number; - error?: string; - warnings?: string[]; - consoleLogCount?: number; - totalRowCount?: number; - sampleRecords?: Record[]; - resultStatistics?: Record; + resultMode: OperatorResultMode; + sampleTuples: [number, Tuple][]; + totalTuplesCount: number; +} + +// Per-operator execution summary returned by the sync-execution backend. +// Orthogonal sub-summaries replace the previous flat `OperatorInfo`. +export interface OperatorExecutionSummary { + state: OperatorState; + // Empty means the operator did not fail. + errorMessages: ReadonlyArray; + // Absent when the operator produced no materialized result. + resultSummary?: OperatorResultSummary; + // Absent when the operator produced no console output. + consoleMessages?: ConsoleMessageSummary[]; +} + +// The result of one synchronous workflow execution. +export interface WorkflowExecutionSummary { + // True only on a clean run; can be false even when state is "Completed" + // (e.g. an operator logged a console error without aborting the run). + success: boolean; + state: WorkflowExecutionState; + operators: Record; + // Workflow-level errors (timeouts, init/compile failures, fatal errors); + // empty means none. For workflow-level failures, operatorId/workerId are empty. + errors: WorkflowFatalError[]; } diff --git a/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala index 3c9da603c86..1ba9b138f6c 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala @@ -22,11 +22,12 @@ package org.apache.texera.web.resource import com.fasterxml.jackson.databind.node.ObjectNode import com.typesafe.scalalogging.LazyLogging import io.dropwizard.auth.Auth +import lombok.Generated import org.apache.texera.common.config.ApplicationConfig import org.apache.texera.amber.core.storage.DocumentFactory import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.core.storage.model.VirtualDocument -import org.apache.texera.amber.core.tuple.Tuple +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} import org.apache.texera.amber.core.virtualidentity.{ ExecutionIdentity, OperatorIdentity, @@ -44,6 +45,12 @@ import org.apache.texera.amber.engine.common.executionruntimestate.{ ExecutionMetadataStore, ExecutionStatsStore } +import org.apache.texera.amber.core.workflowruntimestate.FatalErrorType.{ + COMPILATION_ERROR, + EXECUTION_FAILURE +} +import org.apache.texera.amber.core.workflowruntimestate.{FatalErrorType, WorkflowFatalError} +import com.google.protobuf.timestamp.Timestamp import io.reactivex.rxjava3.core.Observable import org.apache.texera.auth.SessionUser import org.apache.texera.dao.SqlServer @@ -55,6 +62,7 @@ import org.apache.texera.web.service.{ExecutionResultService, WorkflowService} import org.apache.texera.web.storage.ExecutionStateStore.updateWorkflowState import java.net.URI +import java.time.Instant import java.util.concurrent.TimeUnit import javax.annotation.security.RolesAllowed import javax.ws.rs._ @@ -63,6 +71,7 @@ import scala.collection.mutable import scala.jdk.CollectionConverters._ import com.fasterxml.jackson.databind.ObjectMapper +@Generated case class SyncExecutionRequest( executionName: String, logicalPlan: LogicalPlanPojo, @@ -73,40 +82,45 @@ case class SyncExecutionRequest( maxOperatorResultCellCharLimit: Int ) -case class ConsoleMessageInfo( +@Generated +case class ConsoleMessageSummary( msgType: String, title: String, message: String ) -case class PortShape( - portIndex: Int, - rows: Long +@Generated +case class OperatorResultSummary( + resultMode: String, // "table" or "visualization" + // Sampled output rows; each tuple carries its original row index. + sampleTuples: List[(Int, Tuple)], + totalTuplesCount: Int ) -case class OperatorInfo( +// Per-operator execution summary. Orthogonal sub-summaries replace the previous +// flat OperatorInfo; must stay in sync with agent-service's OperatorExecutionSummary. +// `errorMessages` reuses the engine's WorkflowFatalError, the same type the +// compiling service returns for compilation errors, for one consistent wire shape. +@Generated +case class OperatorExecutionSummary( state: String, - inputTuples: Long, - outputTuples: Long, - inputPortShapes: Option[List[PortShape]], - resultMode: String, // "table" or "visualization" - result: Option[Any], // JSON array (List[ObjectNode]) - totalRowCount: Option[Int], - displayedRows: Option[Int], - truncated: Option[Boolean], - consoleLogs: Option[List[ConsoleMessageInfo]], - error: Option[String], - warnings: Option[List[String]] + errorMessages: List[WorkflowFatalError], // empty means the operator did not fail + resultSummary: Option[OperatorResultSummary], + consoleMessages: Option[List[ConsoleMessageSummary]] ) -case class SyncExecutionResult( +@Generated +case class WorkflowExecutionSummary( success: Boolean, state: String, - operators: Map[String, OperatorInfo], - compilationErrors: Option[Map[String, String]], - errors: Option[List[String]] + operators: Map[String, OperatorExecutionSummary], + errors: List[WorkflowFatalError] // empty means none ) +// Internal (not serialized): a sampled row's original index plus its processed/truncated +// JSON node, used while sampling and size-estimating before conversion to an engine Tuple. +private case class SampledRow(rowIndex: Int, node: ObjectNode) + sealed trait TerminationReason case class TerminalStateReached(state: ExecutionMetadataStore) extends TerminationReason case class ConsoleErrorDetected(consoleState: ExecutionConsoleStore) extends TerminationReason @@ -129,7 +143,7 @@ class SyncExecutionResource extends LazyLogging { @PathParam("cuid") computingUnitId: Int, request: SyncExecutionRequest, @Auth user: SessionUser - ): SyncExecutionResult = { + ): WorkflowExecutionSummary = { val timeoutSeconds = request.timeoutSeconds val maxOperatorResultCharLimit = @@ -176,12 +190,11 @@ class SyncExecutionResource extends LazyLogging { val executionService = workflowService.executionService.getValue if (executionService == null) { - return SyncExecutionResult( + return WorkflowExecutionSummary( success = false, state = "Error", operators = Map.empty, - compilationErrors = None, - errors = Some(List("Failed to initialize execution service")) + errors = List(workflowError(EXECUTION_FAILURE, "Failed to initialize execution service")) ) } @@ -254,21 +267,20 @@ class SyncExecutionResource extends LazyLogging { } catch { case _: java.util.concurrent.TimeoutException => killExecution(executionService) - return SyncExecutionResult( + return WorkflowExecutionSummary( success = false, state = "Killed", operators = Map.empty, - compilationErrors = None, - errors = Some(List(s"Timeout after $timeoutSeconds seconds")) + errors = + List(workflowError(EXECUTION_FAILURE, s"Timeout after $timeoutSeconds seconds")) ) case e: Exception => logger.error(s"Error waiting for execution: ${e.getMessage}", e) - return SyncExecutionResult( + return WorkflowExecutionSummary( success = false, state = "Error", operators = Map.empty, - compilationErrors = None, - errors = Some(List(e.getMessage)) + errors = List(workflowError(EXECUTION_FAILURE, messageOrUnknown(e))) ) } } @@ -314,26 +326,11 @@ class SyncExecutionResource extends LazyLogging { inMemoryConsoleState ) - val fatalErrors = finalState.fatalErrors - .map(err => s"${err.`type`}: ${err.message}") - .toList - - val hasOperatorConsoleError = operatorInfos.values.exists(_.error.isDefined) - - val stateString = - if (terminatedByConsoleError) "Failed" - else if (terminatedByTargetResults) "Completed" - else stateToString(finalState.state) - - val isSuccess = (finalState.state == COMPLETED || terminatedByTargetResults) && - !hasOperatorConsoleError && !terminatedByConsoleError - - SyncExecutionResult( - success = isSuccess, - state = stateString, - operators = operatorInfos, - compilationErrors = None, - errors = if (fatalErrors.nonEmpty) Some(fatalErrors) else None + assembleExecutionSummary( + finalState, + operatorInfos, + terminatedByConsoleError, + terminatedByTargetResults ) } catch { @@ -343,6 +340,38 @@ class SyncExecutionResource extends LazyLogging { } } + /** + * Assemble the final workflow execution summary from the terminal metadata state, the + * per-operator summaries, and the termination flags. Extracted from `executeWorkflowSync` + * as a pure function so the success/state derivation can be unit-tested without a live + * engine. Behavior is identical to the inlined version. + */ + private[resource] def assembleExecutionSummary( + finalState: ExecutionMetadataStore, + operatorInfos: Map[String, OperatorExecutionSummary], + terminatedByConsoleError: Boolean, + terminatedByTargetResults: Boolean + ): WorkflowExecutionSummary = { + val fatalErrors = finalState.fatalErrors.toList + + val hasOperatorConsoleError = operatorInfos.values.exists(_.errorMessages.nonEmpty) + + val stateString = + if (terminatedByConsoleError) "Failed" + else if (terminatedByTargetResults) "Completed" + else stateToString(finalState.state) + + val isSuccess = (finalState.state == COMPLETED || terminatedByTargetResults) && + !hasOperatorConsoleError && !terminatedByConsoleError + + WorkflowExecutionSummary( + success = isSuccess, + state = stateString, + operators = operatorInfos, + errors = fatalErrors + ) + } + private def shutdownPreviousExecution(workflowService: WorkflowService): Unit = { try { val previousEs = workflowService.executionService.getValue @@ -381,9 +410,9 @@ class SyncExecutionResource extends LazyLogging { targetOperatorIds: List[String], maxOperatorResultCharLimit: Int, maxOperatorResultCellCharLimit: Int, - inMemoryConsoleState: Option[ExecutionConsoleStore] = None - ): Map[String, OperatorInfo] = { - val operatorInfos = mutable.Map[String, OperatorInfo]() + inMemoryConsoleState: Option[ExecutionConsoleStore] + ): Map[String, OperatorExecutionSummary] = { + val operatorInfos = mutable.Map[String, OperatorExecutionSummary]() val statsState = executionService.executionStateStore.statsStore.getState val operatorStats = statsState.operatorInfo @@ -406,23 +435,9 @@ class SyncExecutionResource extends LazyLogging { for (opId <- targetOps) { val stats = operatorStats.get(opId) - val (state, inputTuples, outputTuples): (String, Long, Long) = stats match { - case Some(s) => - val inputCount = s.operatorStatistics.inputMetrics.map(_.tupleMetrics.count).sum - val outputCount = s.operatorStatistics.outputMetrics.map(_.tupleMetrics.count).sum - (stateToString(s.operatorState), inputCount, outputCount) - case None => ("Unknown", 0L, 0L) - } - - val inputPortShapes: Option[List[PortShape]] = stats - .map { s => - s.operatorStatistics.inputMetrics.map { pm => - PortShape(pm.portId.id, pm.tupleMetrics.count) - }.toList - } - .filter(_.nonEmpty) + val state = stats.map(s => stateToString(s.operatorState)).getOrElse("Unknown") - val (resultMode, result, totalRowCount, displayedRows, truncated) = + val (resultMode, result, totalTuplesCount) = collectOperatorResult( executionId, opId, @@ -439,7 +454,7 @@ class SyncExecutionResource extends LazyLogging { .get(opId) .map { opConsole => opConsole.consoleMessages.map { msg => - ConsoleMessageInfo( + ConsoleMessageSummary( msgType = msg.msgType.name, title = msg.title, message = msg.message @@ -450,40 +465,70 @@ class SyncExecutionResource extends LazyLogging { } } - // Python writes the full error text to `message`; Scala writes it to `title` - // (with a stack trace in `message`). Pick whichever is longer to avoid losing detail. - val errorMsg = consoleLogs.flatMap( - _.find(_.msgType == "ERROR").map { e => - if (e.message.nonEmpty && e.message.length > e.title.length) e.message - else e.title - } + operatorInfos(opId) = buildOperatorExecutionSummary( + opId, + state, + resultMode, + result, + totalTuplesCount, + consoleLogs ) + } - // Convention: PRINT messages prefixed with "WARNING: " surface as warnings. - val warningMsgs = consoleLogs - .map(_.filter(_.title.startsWith("WARNING: ")).map(_.title)) - .filter(_.nonEmpty) + operatorInfos.toMap + } - operatorInfos(opId) = OperatorInfo( - state = state, - inputTuples = inputTuples, - outputTuples = outputTuples, - inputPortShapes = inputPortShapes, + /** + * Build the per-operator execution summary from the operator's state, materialized result, + * and console logs. Extracted from `collectOperatorInfos` as a pure function so the + * error-extraction and sub-summary wiring can be unit-tested directly. Behavior is + * identical to the inlined version. + */ + private[resource] def buildOperatorExecutionSummary( + opId: String, + state: String, + resultMode: String, + result: Option[List[SampledRow]], + totalTuplesCount: Option[Int], + consoleLogs: Option[List[ConsoleMessageSummary]] + ): OperatorExecutionSummary = { + // Python writes the full error text to `message`; Scala writes it to `title` + // (with a stack trace in `message`). Pick whichever is longer to avoid losing detail. + val errorMsg = consoleLogs.flatMap( + _.find(_.msgType == "ERROR").map { e => + if (e.message.nonEmpty && e.message.length > e.title.length) e.message + else e.title + } + ) + + // Absent when the operator produced no materialized result. `result` and + // `totalTuplesCount` are populated together, so map over the former. + val resultSummary = result.map { rows => + OperatorResultSummary( resultMode = resultMode, - result = result, - totalRowCount = totalRowCount, - displayedRows = displayedRows, - truncated = truncated, - consoleLogs = consoleLogs, - error = errorMsg, - warnings = warningMsgs + sampleTuples = rows.map(r => (r.rowIndex, toStringTuple(r.node))), + totalTuplesCount = totalTuplesCount.getOrElse(0) ) } - operatorInfos.toMap + // Per-operator runtime errors come from console ERROR logs; surface them as + // EXECUTION_FAILURE WorkflowFatalErrors (same type the compiler emits for + // COMPILATION_ERRORs). Empty list means the operator did not fail. + val errorMessages = errorMsg + .map(msg => + List(WorkflowFatalError(EXECUTION_FAILURE, Timestamp(Instant.now), msg, "", opId)) + ) + .getOrElse(List.empty) + + OperatorExecutionSummary( + state = state, + errorMessages = errorMessages, + resultSummary = resultSummary, + consoleMessages = consoleLogs + ) } - private def handleExecutionError(e: Exception): SyncExecutionResult = { + private def handleExecutionError(e: Exception): WorkflowExecutionSummary = { val errorMsg = e.getMessage val isCompilationError = errorMsg != null && ( errorMsg.contains("compilation") || @@ -493,24 +538,46 @@ class SyncExecutionResource extends LazyLogging { ) if (isCompilationError) { - SyncExecutionResult( + WorkflowExecutionSummary( success = false, state = "CompilationFailed", operators = Map.empty, - compilationErrors = Some(Map("error" -> errorMsg)), - errors = Some(List(errorMsg)) + errors = List(workflowError(COMPILATION_ERROR, errorMsg)) ) } else { - SyncExecutionResult( + WorkflowExecutionSummary( success = false, state = "Error", operators = Map.empty, - compilationErrors = None, - errors = Some(List(Option(e.getMessage).getOrElse("Unknown error"))) + errors = List(workflowError(EXECUTION_FAILURE, messageOrUnknown(e))) ) } } + private def workflowError(errorType: FatalErrorType, message: String): WorkflowFatalError = + WorkflowFatalError(errorType, Timestamp(Instant.now), message, "", "", "") + + private def messageOrUnknown(e: Exception): String = + Option(e.getMessage).getOrElse("Unknown error") + + // Wrap a processed/truncated JSON row in an engine Tuple. Cell truncation replaces typed + // values (e.g. binary) with display strings that no longer match the operator's real column + // types, so each row gets a synthetic all-STRING schema over its columns. Null stays null; + // container values are emitted as their JSON text. + private def toStringTuple(node: ObjectNode): Tuple = { + val fieldNames = node.fieldNames().asScala.toList + val schema = Schema(fieldNames.map(name => new Attribute(name, AttributeType.STRING))) + val fields: Array[Any] = fieldNames.map { name => + node.get(name) match { + case null => null + case v if v.isNull => null + case v if v.isValueNode => v.asText() + case v => v.toString + } + }.toArray + Tuple(schema, fields) + } + /** * Symmetric truncation: fill half the char budget from the front of the result, keep a * sliding-window of the most recent tuples for the back half. Returns a JSON array; @@ -521,9 +588,7 @@ class SyncExecutionResource extends LazyLogging { opId: String, maxOperatorResultCharLimit: Int, maxOperatorResultCellCharLimit: Int - ): (String, Option[Any], Option[Int], Option[Int], Option[Boolean]) = { - import com.fasterxml.jackson.databind.node.ObjectNode - + ): (String, Option[List[SampledRow]], Option[Int]) = { try { val storageUriOption = WorkflowExecutionsResource.getResultUriByLogicalPortId( executionId, @@ -531,171 +596,154 @@ class SyncExecutionResource extends LazyLogging { PortIdentity() ) - storageUriOption match { - case Some(storageUri) => - val document = DocumentFactory - .openDocument(storageUri) - ._1 - .asInstanceOf[VirtualDocument[Tuple]] - - val totalCount = document.getCount.toInt - val mapper = new ObjectMapper() - val tupleIterator = document.get() - - if (totalCount == 0 || !tupleIterator.hasNext) { - return ( - "table", - Some(List.empty[ObjectNode].asJava), - Some(0), - Some(0), - Some(false) - ) - } + if (storageUriOption.isEmpty) { + ("table", None, None) + } else { + val document = DocumentFactory + .openDocument(storageUriOption.get) + ._1 + .asInstanceOf[VirtualDocument[Tuple]] - // A single tuple with html-content / json-content is a visualization payload — - // the frontend renders it as an iframe rather than a table. - val firstTuple = tupleIterator.next() - if (totalCount == 1 && isVisualizationTuple(firstTuple)) { - val jsonResults = - ExecutionResultService.convertTuplesToJson(List(firstTuple), isVisualization = true) - jsonResults.foreach( - _.asInstanceOf[ObjectNode].put("__is_visualization__", true) - ) - return ( - "visualization", - Some(jsonResults), - Some(totalCount), - Some(1), - Some(false) - ) - } + sampleAndTruncateTuples( + document.get(), + document.getCount.toInt, + maxOperatorResultCharLimit, + maxOperatorResultCellCharLimit + ) + } + } catch { + case e: Exception => + logger.warn(s"Error collecting result for operator $opId", e) + ("table", None, None) + } + } - // __row_index__ preserves the original position so the frontend can show - // "row N" correctly after symmetric truncation drops the middle. - var rowIndex = 0 - val firstJson = ExecutionResultService.convertTuplesToJson(List(firstTuple)).head - val truncatedFirst = truncateSingleTuple(firstJson, maxOperatorResultCellCharLimit) - truncatedFirst.put("__row_index__", rowIndex) - val firstSize = estimateTupleSize(truncatedFirst, mapper) - - if (firstSize >= maxOperatorResultCharLimit) { - return ( - "table", - Some(List(truncatedFirst).asJava), - Some(totalCount), - Some(1), - Some(true) - ) - } + /** + * Sample and symmetrically truncate the tuples of a materialized result so the payload + * stays within `maxOperatorResultCharLimit`. Extracted from `collectOperatorResult` as a + * pure function over the tuple iterator so the sampling/truncation branches can be + * unit-tested without a storage backend. Behavior is identical to the inlined version. + */ + private[resource] def sampleAndTruncateTuples( + tupleIterator: Iterator[Tuple], + totalCount: Int, + maxOperatorResultCharLimit: Int, + maxOperatorResultCellCharLimit: Int + ): (String, Option[List[SampledRow]], Option[Int]) = { + val mapper = new ObjectMapper() - val halfLimit = maxOperatorResultCharLimit / 2 - val truncationNoticeSize = 50 // reserved for the "...skipped..." marker - - val frontTuples = mutable.ListBuffer[ObjectNode](truncatedFirst) - var frontSize = firstSize - var processedCount = 1 - - while (tupleIterator.hasNext && frontSize < halfLimit) { - val tuple = tupleIterator.next() - rowIndex += 1 - processedCount += 1 - val jsonTuple = ExecutionResultService.convertTuplesToJson(List(tuple)).head - val truncatedTuple = truncateSingleTuple(jsonTuple, maxOperatorResultCellCharLimit) - truncatedTuple.put("__row_index__", rowIndex) - val tupleSize = estimateTupleSize(truncatedTuple, mapper) - - if (frontSize + tupleSize <= halfLimit) { - frontTuples += truncatedTuple - frontSize += tupleSize - } else { - // Front is full — switch to a sliding window for the back half. - val backBuffer = mutable.ArrayBuffer[(ObjectNode, Int)]() - backBuffer += ((truncatedTuple, tupleSize)) - var backSize = tupleSize - - while (tupleIterator.hasNext) { - val t = tupleIterator.next() - rowIndex += 1 - processedCount += 1 - val jt = ExecutionResultService.convertTuplesToJson(List(t)).head - val tt = truncateSingleTuple(jt, maxOperatorResultCellCharLimit) - tt.put("__row_index__", rowIndex) - val ts = estimateTupleSize(tt, mapper) - - backBuffer += ((tt, ts)) - backSize += ts - - while (backSize > halfLimit - truncationNoticeSize && backBuffer.size > 1) { - val (_, removedSize) = backBuffer.remove(0) - backSize -= removedSize - } - } - - val backTuples = backBuffer.map(_._1).toList - val allTuples = frontTuples.toList ++ backTuples - val skippedRows = totalCount - allTuples.size - - return ( - "table", - Some(allTuples.asJava), - Some(totalCount), - Some(allTuples.size), - Some(skippedRows > 0) - ) - } - } + if (totalCount == 0 || !tupleIterator.hasNext) { + return ("table", Some(List.empty[SampledRow]), Some(0)) + } - if (tupleIterator.hasNext) { - val backBuffer = mutable.ArrayBuffer[(ObjectNode, Int)]() - var backSize = 0 - - while (tupleIterator.hasNext) { - val t = tupleIterator.next() - rowIndex += 1 - processedCount += 1 - val jt = ExecutionResultService.convertTuplesToJson(List(t)).head - val tt = truncateSingleTuple(jt, maxOperatorResultCellCharLimit) - tt.put("__row_index__", rowIndex) - val ts = estimateTupleSize(tt, mapper) - - backBuffer += ((tt, ts)) - backSize += ts - - while (backSize > halfLimit - truncationNoticeSize && backBuffer.size > 1) { - val (_, removedSize) = backBuffer.remove(0) - backSize -= removedSize - } - } + // A single tuple with html-content / json-content is a visualization payload — + // the frontend renders it as an iframe rather than a table. + val firstTuple = tupleIterator.next() + if (totalCount == 1 && isVisualizationTuple(firstTuple)) { + val jsonResults = + ExecutionResultService.convertTuplesToJson(List(firstTuple), isVisualization = true) + val rows = jsonResults.zipWithIndex.map { case (json, idx) => SampledRow(idx, json) } + return ("visualization", Some(rows), Some(totalCount)) + } - val backTuples = backBuffer.map(_._1).toList - val allTuples = frontTuples.toList ++ backTuples - val skippedRows = totalCount - allTuples.size + // rowIndex preserves the original position so the client can show "row N" + // correctly after symmetric truncation drops the middle. + var rowIndex = 0 + val firstJson = ExecutionResultService.convertTuplesToJson(List(firstTuple)).head + val truncatedFirst = truncateSingleTuple(firstJson, maxOperatorResultCellCharLimit) + val firstSize = estimateTupleSize(truncatedFirst, mapper) - ( - "table", - Some(allTuples.asJava), - Some(totalCount), - Some(allTuples.size), - Some(skippedRows > 0) - ) - } else { - ( - "table", - Some(frontTuples.toList.asJava), - Some(totalCount), - Some(frontTuples.size), - Some(false) - ) - } + if (firstSize >= maxOperatorResultCharLimit) { + return ("table", Some(List(SampledRow(rowIndex, truncatedFirst))), Some(totalCount)) + } + + val halfLimit = maxOperatorResultCharLimit / 2 + val truncationNoticeSize = 50 // reserved for the "...skipped..." marker + + val frontRows = mutable.ListBuffer[SampledRow](SampledRow(rowIndex, truncatedFirst)) + var frontSize = firstSize - case None => - ("table", None, None, None, None) + while (tupleIterator.hasNext && frontSize < halfLimit) { + val tuple = tupleIterator.next() + rowIndex += 1 + val jsonTuple = ExecutionResultService.convertTuplesToJson(List(tuple)).head + val truncatedTuple = truncateSingleTuple(jsonTuple, maxOperatorResultCellCharLimit) + val tupleSize = estimateTupleSize(truncatedTuple, mapper) + val row = SampledRow(rowIndex, truncatedTuple) + + if (frontSize + tupleSize <= halfLimit) { + frontRows += row + frontSize += tupleSize + } else { + // Front is full — switch to a sliding window for the back half, seeded with the + // row that overflowed the front. + val backRows = collectBackWindow( + tupleIterator, + rowIndex, + Some((row, tupleSize)), + halfLimit - truncationNoticeSize, + maxOperatorResultCellCharLimit, + mapper + ) + return ("table", Some(frontRows.toList ++ backRows), Some(totalCount)) } - } catch { - case e: Exception => - logger.warn(s"Error collecting result for operator $opId: ${e.getMessage}", e) - ("table", None, None, None, None) } + + if (tupleIterator.hasNext) { + val backRows = collectBackWindow( + tupleIterator, + rowIndex, + None, + halfLimit - truncationNoticeSize, + maxOperatorResultCellCharLimit, + mapper + ) + ("table", Some(frontRows.toList ++ backRows), Some(totalCount)) + } else { + ("table", Some(frontRows.toList), Some(totalCount)) + } + } + + /** + * Consume the remaining tuples into a sliding back-window that keeps the most recent rows + * within `backSizeLimit` estimated characters, always retaining at least one row. `seed` + * is the already-processed row (and its size) that overflowed the front half, if any. + * Shared by the "front already full" and trailing back-window branches of + * `sampleAndTruncateTuples`. + */ + private def collectBackWindow( + tupleIterator: Iterator[Tuple], + startRowIndex: Int, + seed: Option[(SampledRow, Int)], + backSizeLimit: Int, + maxOperatorResultCellCharLimit: Int, + mapper: ObjectMapper + ): List[SampledRow] = { + val backBuffer = mutable.ArrayBuffer[(SampledRow, Int)]() + var backSize = 0 + seed.foreach { + case (row, size) => + backBuffer += ((row, size)) + backSize += size + } + + var rowIndex = startRowIndex + while (tupleIterator.hasNext) { + val t = tupleIterator.next() + rowIndex += 1 + val jt = ExecutionResultService.convertTuplesToJson(List(t)).head + val tt = truncateSingleTuple(jt, maxOperatorResultCellCharLimit) + val ts = estimateTupleSize(tt, mapper) + + backBuffer += ((SampledRow(rowIndex, tt), ts)) + backSize += ts + + while (backSize > backSizeLimit && backBuffer.size > 1) { + backSize -= backBuffer.remove(0)._2 + } + } + + backBuffer.map(_._1).toList } private def truncateSingleTuple( @@ -762,7 +810,7 @@ class SyncExecutionResource extends LazyLogging { private def collectConsoleLogs( executionId: ExecutionIdentity, opId: String - ): Option[List[ConsoleMessageInfo]] = { + ): Option[List[ConsoleMessageSummary]] = { try { val uriOption = getConsoleMessageUri(executionId, OperatorIdentity(opId)) @@ -777,7 +825,7 @@ class SyncExecutionResource extends LazyLogging { val protoString = tuple.getField[String](0) val msg = ConsoleMessage.fromAscii(protoString) Some( - ConsoleMessageInfo( + ConsoleMessageSummary( msgType = msg.msgType.name, title = msg.title, message = msg.message diff --git a/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala new file mode 100644 index 00000000000..56fec7e484b --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala @@ -0,0 +1,950 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.web.resource + +import com.fasterxml.jackson.databind.ObjectMapper +import com.fasterxml.jackson.module.scala.DefaultScalaModule +import com.google.protobuf.timestamp.Timestamp +import org.apache.texera.amber.core.storage.{DocumentFactory, VFSURIFactory} +import org.apache.texera.amber.core.storage.model.{BufferedItemWriter, VirtualDocument} +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.virtualidentity.{ + ExecutionIdentity, + OperatorIdentity, + PhysicalOpIdentity, + WorkflowIdentity +} +import org.apache.texera.amber.core.workflow.{ + GlobalPortIdentity, + PortIdentity, + WorkflowContext, + WorkflowSettings +} +import org.apache.texera.amber.core.workflowruntimestate.FatalErrorType.{ + COMPILATION_ERROR, + EXECUTION_FAILURE +} +import org.apache.texera.amber.core.workflowruntimestate.WorkflowFatalError +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + ConsoleMessage, + ConsoleMessageType +} +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState +import org.apache.texera.amber.engine.common.executionruntimestate.{ + ExecutionConsoleStore, + ExecutionMetadataStore, + ExecutionStatsStore, + OperatorMetrics, + OperatorStatistics +} +import org.apache.texera.amber.util.serde.GlobalPortIdentitySerde.SerdeOps +import org.apache.texera.auth.SessionUser +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.tables.daos.{ + UserDao, + WorkflowDao, + WorkflowExecutionsDao, + WorkflowVersionDao +} +import org.apache.texera.dao.jooq.generated.tables.pojos.{ + User, + Workflow, + WorkflowExecutions, + WorkflowVersion +} +import org.apache.texera.dao.jooq.generated.Tables.OPERATOR_PORT_EXECUTIONS +import org.apache.texera.web.model.websocket.request.{LogicalPlanPojo, WorkflowExecuteRequest} +import org.apache.texera.web.storage.ExecutionStateStore +import org.apache.texera.web.service.{ConsoleMessageProcessor, WorkflowService} +import org.scalatest.{BeforeAndAfterAll, PrivateMethodTester} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.net.URI +import java.sql.{Timestamp => SqlTimestamp} +import java.time.Instant +import java.util.concurrent.ConcurrentHashMap +import java.util.UUID + +/** + * Unit tests for the pure parts of [[SyncExecutionResource]] that this PR introduced: the + * `handleExecutionError` error-classification branch, and the behavior-preserving + * extract-method helpers `sampleAndTruncateTuples` (the result sampling/truncation logic + * lifted out of `collectOperatorResult`), `buildOperatorExecutionSummary` (the + * per-operator summary construction lifted out of `collectOperatorInfos`), and + * `assembleExecutionSummary` (the success/state derivation lifted out of + * `executeWorkflowSync`). + * + * The remaining changed code paths (`executeWorkflowSync`, and the storage/DB plumbing in + * `collectOperatorResult` / `collectOperatorInfos` around these helpers) require a live + * Pekko execution engine, an Iceberg-backed result store, and DB-persisted port executions, + * so they are exercised by integration tests rather than here. + */ +class SyncExecutionResourceSpec + extends AnyFlatSpec + with Matchers + with PrivateMethodTester + with BeforeAndAfterAll + with MockTexeraDB { + + private val mapper = new ObjectMapper() + private val resource = new SyncExecutionResource + private var nextDbId = 900000 + + override protected def beforeAll(): Unit = { + initializeDBAndReplaceDSLContext() + } + + override protected def afterAll(): Unit = { + shutdownDB() + } + + private def sampleRow(idx: Int, k: String, v: String): SampledRow = { + val node = mapper.createObjectNode() + node.put(k, v) + SampledRow(rowIndex = idx, node = node) + } + + private def nextId(): Int = { + nextDbId += 1 + nextDbId + } + + private def insertExecutionRow(): ExecutionIdentity = { + val now = new SqlTimestamp(System.currentTimeMillis()) + val suffix = UUID.randomUUID().toString.substring(0, 8) + val uid = nextId() + val wid = nextId() + val vid = nextId() + val eid = nextId() + + val user = new User + user.setUid(uid) + user.setName(s"sync-user-$suffix") + user.setEmail(s"sync-user-$suffix@example.com") + user.setPassword("password") + new UserDao(getDSLContext.configuration()).insert(user) + + val workflow = new Workflow + workflow.setWid(wid) + workflow.setName(s"sync-workflow-$suffix") + workflow.setContent("{}") + workflow.setDescription("") + workflow.setCreationTime(now) + workflow.setLastModifiedTime(now) + new WorkflowDao(getDSLContext.configuration()).insert(workflow) + + val version = new WorkflowVersion + version.setVid(vid) + version.setWid(wid) + version.setContent("{}") + version.setCreationTime(now) + new WorkflowVersionDao(getDSLContext.configuration()).insert(version) + + val execution = new WorkflowExecutions + execution.setEid(eid) + execution.setVid(vid) + execution.setUid(uid) + execution.setStatus(0.toByte) + execution.setName(s"sync-execution-$suffix") + execution.setEnvironmentVersion("test engine") + execution.setStartingTime(now) + new WorkflowExecutionsDao(getDSLContext.configuration()).insert(execution) + + ExecutionIdentity(eid.toLong) + } + + private def buildExecutionService( + stateStore: ExecutionStateStore + ): org.apache.texera.web.service.WorkflowExecutionService = { + val request = WorkflowExecuteRequest( + executionName = "sync-test", + engineVersion = "test", + logicalPlan = LogicalPlanPojo(List.empty, List.empty, List.empty, List.empty), + replayFromExecution = None, + workflowSettings = WorkflowSettings(), + emailNotificationEnabled = false, + computingUnitId = 0 + ) + new org.apache.texera.web.service.WorkflowExecutionService( + null, + new WorkflowContext(), + null, + request, + stateStore, + (_: Throwable) => (), + None, + new URI("vfs:///sync-resource-test") + ) + } + + private class StubWorkflowService( + workflowId: WorkflowIdentity, + computingUnitId: Int, + initHook: StubWorkflowService => Unit + ) extends WorkflowService(workflowId, computingUnitId, 1) { + override def initExecutionService( + req: WorkflowExecuteRequest, + userOpt: Option[User], + sessionUri: URI + ): Unit = initHook(this) + } + + private def workflowServiceMapping: ConcurrentHashMap[String, WorkflowService] = { + val accessor = WorkflowService.getClass.getDeclaredMethod( + "org$apache$texera$web$service$WorkflowService$$workflowServiceMapping" + ) + accessor.setAccessible(true) + accessor.invoke(WorkflowService).asInstanceOf[ConcurrentHashMap[String, WorkflowService]] + } + + private def withRegisteredWorkflowService( + service: WorkflowService + )(body: => WorkflowExecutionSummary): WorkflowExecutionSummary = { + val key = WorkflowService.mkWorkflowStateId(service.workflowId) + workflowServiceMapping.put(key, service) + try { + body + } finally { + workflowServiceMapping.remove(key) + service.unsubscribeAll() + } + } + + private def sessionUser(): SessionUser = { + val user = new User + user.setUid(nextId()) + user.setName("sync-session-user") + user.setEmail("sync-session-user@example.com") + new SessionUser(user) + } + + private def syncRequest(timeoutSeconds: Int = 1): SyncExecutionRequest = + SyncExecutionRequest( + executionName = "sync-test", + logicalPlan = LogicalPlanPojo(List.empty, List.empty, List.empty, List.empty), + workflowSettings = None, + targetOperatorIds = List.empty, + timeoutSeconds = timeoutSeconds, + maxOperatorResultCharLimit = 100000, + maxOperatorResultCellCharLimit = 100000 + ) + + private def runWithStubWorkflow( + initHook: StubWorkflowService => Unit, + request: SyncExecutionRequest = syncRequest() + ): WorkflowExecutionSummary = { + val workflowId = WorkflowIdentity(nextId().toLong) + val service = new StubWorkflowService(workflowId, computingUnitId = 0, initHook) + withRegisteredWorkflowService(service) { + resource.executeWorkflowSync(workflowId.id, service.computingUnitId, request, sessionUser()) + } + } + + private def failMetadataObservable(stateStore: ExecutionStateStore, error: Throwable): Unit = { + val subjectField = stateStore.metadataStore.getClass.getDeclaredField("serializedSubject") + subjectField.setAccessible(true) + subjectField + .get(stateStore.metadataStore) + .asInstanceOf[io.reactivex.rxjava3.subjects.Subject[ExecutionMetadataStore]] + .onError(error) + } + + // handleExecutionError is private; reflectively invoke it (no production change needed). + private val handleExecutionError = + PrivateMethod[WorkflowExecutionSummary](Symbol("handleExecutionError")) + + private val collectOperatorInfos = + PrivateMethod[Map[String, OperatorExecutionSummary]](Symbol("collectOperatorInfos")) + + private val collectOperatorResult = + PrivateMethod[(String, Option[List[SampledRow]], Option[Int])](Symbol("collectOperatorResult")) + + private val symmetricTruncateCellValue = + PrivateMethod[String](Symbol("symmetricTruncateCellValue")) + + private def classify(message: String): WorkflowExecutionSummary = + resource invokePrivate handleExecutionError(new RuntimeException(message)) + + "handleExecutionError" should "classify lowercase 'compilation' messages as CompilationFailed" in { + val summary = classify("compilation failed for the plan") + summary.success shouldBe false + summary.state shouldBe "CompilationFailed" + summary.operators shouldBe empty + summary.errors should have size 1 + summary.errors.head.`type` shouldBe COMPILATION_ERROR + summary.errors.head.message shouldBe "compilation failed for the plan" + } + + it should "classify unrecognized messages as a generic Error" in { + val summary = classify("something unexpected happened") + summary.success shouldBe false + summary.state shouldBe "Error" + summary.operators shouldBe empty + summary.errors should have size 1 + summary.errors.head.`type` shouldBe EXECUTION_FAILURE + summary.errors.head.message shouldBe "something unexpected happened" + } + + it should "fall back to 'Unknown error' when the exception has a null message" in { + val summary = resource invokePrivate handleExecutionError( + new RuntimeException(null.asInstanceOf[String]) + ) + summary.state shouldBe "Error" + summary.errors should have size 1 + summary.errors.head.`type` shouldBe EXECUTION_FAILURE + summary.errors.head.message shouldBe "Unknown error" + } + + // --- sampleAndTruncateTuples (extracted from collectOperatorResult) --------------------- + + private val tableSchema = new Schema(List(new Attribute("col", AttributeType.STRING))) + private def tableTuple(v: String): Tuple = + Tuple.builder(tableSchema).add("col", AttributeType.STRING, v).build() + + private val mixedSchema = new Schema( + List( + new Attribute("col", AttributeType.STRING), + new Attribute("number", AttributeType.INTEGER) + ) + ) + private def mixedTuple(v: String, number: Int): Tuple = + Tuple + .builder(mixedSchema) + .add("col", AttributeType.STRING, v) + .add("number", AttributeType.INTEGER, number) + .build() + + private def materializeResult( + executionId: ExecutionIdentity, + opId: String, + tuples: List[Tuple] + ): VirtualDocument[Tuple] = { + val operatorId = OperatorIdentity(opId) + val globalPort = GlobalPortIdentity( + PhysicalOpIdentity(operatorId, "main"), + PortIdentity(), + input = false + ) + val uri = VFSURIFactory.resultURI( + VFSURIFactory.createPortBaseURI(WorkflowIdentity(1L), executionId, globalPort) + ) + val document = + DocumentFactory.createDocument(uri, tableSchema).asInstanceOf[VirtualDocument[Tuple]] + val writer = + document + .writer(s"sync-result-${executionId.id}-$opId") + .asInstanceOf[BufferedItemWriter[Tuple]] + writer.open() + tuples.foreach(writer.putOne) + writer.close() + getDSLContext + .insertInto(OPERATOR_PORT_EXECUTIONS) + .columns( + OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID, + OPERATOR_PORT_EXECUTIONS.GLOBAL_PORT_ID, + OPERATOR_PORT_EXECUTIONS.RESULT_URI + ) + .values(executionId.id.toInt, globalPort.serializeAsString, uri.toString) + .execute() + document + } + + "collectOperatorResult" should "return no result summary when no result URI exists" in { + val summary = resource invokePrivate collectOperatorResult( + insertExecutionRow(), + "missing-result-op", + 100000, + 100000 + ) + summary shouldBe (("table", None, None)) + } + + it should "sample rows from a materialized result document" in { + val executionId = insertExecutionRow() + val document = + materializeResult(executionId, "result-op", List(tableTuple("a"), tableTuple("b"))) + try { + val (mode, rows, total) = resource invokePrivate collectOperatorResult( + executionId, + "result-op", + 100000, + 100000 + ) + mode shouldBe "table" + rows.get.map(_.rowIndex) shouldBe List(0, 1) + rows.get.map(_.node.get("col").asText()) shouldBe List("a", "b") + total shouldBe Some(2) + } finally { + document.clear() + } + } + + it should "fall back to no result summary when result lookup throws" in { + val summary = resource invokePrivate collectOperatorResult( + null.asInstanceOf[ExecutionIdentity], + "bad-result-op", + 100000, + 100000 + ) + summary shouldBe (("table", None, None)) + + val initializedLoggerMethod = + classOf[SyncExecutionResource].getDeclaredMethod("logger$lzycompute") + initializedLoggerMethod.setAccessible(true) + initializedLoggerMethod.invoke(resource) + + val secondSummary = resource invokePrivate collectOperatorResult( + null.asInstanceOf[ExecutionIdentity], + "bad-result-op-again", + 100000, + 100000 + ) + secondSummary shouldBe (("table", None, None)) + + val loggerField = classOf[SyncExecutionResource].getDeclaredField("logger") + val loggerInitializedField = classOf[SyncExecutionResource].getDeclaredField("bitmap$trans$0") + loggerField.setAccessible(true) + loggerInitializedField.setAccessible(true) + val originalLogger = loggerField.get(resource) + val originalLoggerInitialized = loggerInitializedField.getBoolean(resource) + try { + loggerField.set( + resource, + com.typesafe.scalalogging.Logger(org.slf4j.helpers.NOPLogger.NOP_LOGGER) + ) + loggerInitializedField.setBoolean(resource, true) + val disabledLoggerSummary = resource invokePrivate collectOperatorResult( + null.asInstanceOf[ExecutionIdentity], + "bad-result-op-disabled-logger", + 100000, + 100000 + ) + disabledLoggerSummary shouldBe (("table", None, None)) + } finally { + loggerField.set(resource, originalLogger) + loggerInitializedField.setBoolean(resource, originalLoggerInitialized) + } + } + + "sampleAndTruncateTuples" should "report an empty table for a zero-count / empty iterator" in { + val (mode, rows, total) = + resource.sampleAndTruncateTuples(Iterator.empty, 0, 100000, 100000) + mode shouldBe "table" + rows shouldBe Some(List.empty[SampledRow]) + total shouldBe Some(0) + } + + it should "report an empty table when the iterator is empty despite a positive count" in { + // Exercises the `!tupleIterator.hasNext` half of the guard (count > 0, no rows). + val (mode, rows, total) = + resource.sampleAndTruncateTuples(Iterator.empty, 5, 100000, 100000) + mode shouldBe "table" + rows shouldBe Some(List.empty[SampledRow]) + total shouldBe Some(0) + } + + it should "return visualization mode for a single visualization tuple" in { + val vizSchema = new Schema(List(new Attribute("html-content", AttributeType.STRING))) + val vizTuple = + Tuple.builder(vizSchema).add("html-content", AttributeType.STRING, "
viz
").build() + + val (mode, rows, total) = + resource.sampleAndTruncateTuples(Iterator(vizTuple), 1, 100000, 100000) + mode shouldBe "visualization" + rows.get should have size 1 + rows.get.head.rowIndex shouldBe 0 + total shouldBe Some(1) + } + + it should "return every row untruncated when the whole table fits (front-only path)" in { + val tuples = List(tableTuple("a"), tableTuple("b"), tableTuple("c")) + val (mode, rows, total) = + resource.sampleAndTruncateTuples(tuples.iterator, tuples.size, 100000, 100000) + mode shouldBe "table" + rows.get should have size 3 + rows.get.map(_.rowIndex) shouldBe List(0, 1, 2) + rows.get.map(_.node.get("col").asText()) shouldBe List("a", "b", "c") + total shouldBe Some(3) + } + + it should "truncate oversized text cells while preserving non-text fields" in { + val (mode, rows, total) = + resource.sampleAndTruncateTuples(Iterator(mixedTuple("abcdefghij", 42)), 1, 100000, 8) + mode shouldBe "table" + total shouldBe Some(1) + rows.get.head.node.get("col").asText() shouldBe "abcdefgh" + rows.get.head.node.get("number").asInt() shouldBe 42 + } + + it should "symmetrically truncate cells when enough room remains for both sides" in { + val unchanged = resource invokePrivate symmetricTruncateCellValue("short", 10) + val truncated = resource invokePrivate symmetricTruncateCellValue( + "abcdefghijklmnopqrstuvwxyz", + 21 + ) + + unchanged shouldBe "short" + truncated shouldBe "ab...[truncated]...yz" + } + + it should "keep only the first (truncated) row when a single tuple exceeds the char limit" in { + // Tiny char limit so the first tuple's estimated size already exceeds it. The single + // tuple is not a visualization tuple, so the oversized-first early return fires. + val (mode, rows, total) = + resource.sampleAndTruncateTuples(Iterator(tableTuple("hello world")), 1, 10, 100000) + mode shouldBe "table" + rows.get should have size 1 + rows.get.head.rowIndex shouldBe 0 + total shouldBe Some(1) + } + + it should "drop the middle rows when the front fills and a sliding back-window runs" in { + // ~31 chars/tuple, halfLimit = 100: the front fills after a few rows, then the inner + // else switches to the sliding back-window and drops the middle rows. + val tuples = (0 until 12).map(i => tableTuple("v" * 20)).toList + val (mode, rows, total) = + resource.sampleAndTruncateTuples(tuples.iterator, tuples.size, 200, 100000) + mode shouldBe "table" + total shouldBe Some(12) + rows.get.size should be < 12 + // Front rows keep their original positions; the tail keeps the most recent rows. + rows.get.head.rowIndex shouldBe 0 + rows.get.last.rowIndex shouldBe 11 + } + + it should "run the trailing back-window when the first row alone fills the front half" in { + // First tuple's string is truncated by convertTuplesToJson to 103 chars, giving an + // estimated size >= halfLimit (100) but < the char limit (200), so the front while-loop + // never runs and the trailing back-window path handles the remaining rows. + val tuples = tableTuple("x" * 150) :: (0 until 5).map(_ => tableTuple("y" * 20)).toList + val (mode, rows, total) = + resource.sampleAndTruncateTuples(tuples.iterator, tuples.size, 200, 100000) + mode shouldBe "table" + total shouldBe Some(6) + rows.get.size should be < 6 + rows.get.head.rowIndex shouldBe 0 + rows.get.last.rowIndex shouldBe 5 + } + + it should "keep one oversized row in the trailing back-window" in { + // The back-window limit is smaller than the first tail row. The sliding-window loop still + // retains one row instead of dropping every row from the back sample. + val tuples = List(tableTuple("x" * 150), tableTuple("y" * 80)) + val (mode, rows, total) = + resource.sampleAndTruncateTuples(tuples.iterator, tuples.size, 200, 100000) + mode shouldBe "table" + total shouldBe Some(2) + rows.get.map(_.rowIndex) shouldBe List(0, 1) + } + + // --- buildOperatorExecutionSummary (extracted from collectOperatorInfos) ---------------- + + "buildOperatorExecutionSummary" should "wire a result summary and no errors when only a result is present" in { + val rows = List(sampleRow(0, "col", "v")) + val summary = resource.buildOperatorExecutionSummary( + opId = "op-1", + state = "Completed", + resultMode = "table", + result = Some(rows), + totalTuplesCount = Some(7), + consoleLogs = None + ) + summary.state shouldBe "Completed" + summary.errorMessages shouldBe empty + summary.consoleMessages shouldBe None + summary.resultSummary.get.resultMode shouldBe "table" + // sampled rows are emitted as (originalRowIndex, all-STRING Tuple) pairs + val sampled = summary.resultSummary.get.sampleTuples + sampled.map(_._1) shouldBe List(0) + sampled.head._2.getField[String]("col") shouldBe "v" + sampled.head._2.getSchema.getAttribute("col").getType shouldBe AttributeType.STRING + summary.resultSummary.get.totalTuplesCount shouldBe 7 + } + + // Locks the wire contract the agent-service / frontend consumers depend on: each sampled + // row serializes as a 2-element array [index, {schema:{attributes:[...]}, fields:[...]}]. + it should "serialize sampled rows as [index, {schema, fields}]" in { + val scalaMapper = new ObjectMapper().registerModule(DefaultScalaModule) + val tuple = + Tuple(Schema(List(new Attribute("col", AttributeType.STRING))), Array[Any]("v")) + val summary = + OperatorResultSummary( + resultMode = "table", + sampleTuples = List((0, tuple)), + totalTuplesCount = 1 + ) + val firstRow = + scalaMapper.readTree(scalaMapper.writeValueAsString(summary)).get("sampleTuples").get(0) + firstRow.get(0).asInt() shouldBe 0 + firstRow.get(1).get("fields").get(0).asText() shouldBe "v" + val attr = firstRow.get(1).get("schema").get("attributes").get(0) + attr.get("attributeName").asText() shouldBe "col" + attr.get("attributeType").asText() shouldBe "string" + } + + it should "default a present result summary to zero tuples when the count is absent" in { + val rows = List(sampleRow(0, "col", "v")) + val summary = resource.buildOperatorExecutionSummary( + opId = "op-1", + state = "Completed", + resultMode = "table", + result = Some(rows), + totalTuplesCount = None, + consoleLogs = None + ) + summary.resultSummary.get.totalTuplesCount shouldBe 0 + } + + it should "surface a console ERROR as one EXECUTION_FAILURE error using the longer of title/message" in { + val logs = List( + ConsoleMessageSummary(msgType = "PRINT", title = "noise", message = "ignored"), + ConsoleMessageSummary(msgType = "ERROR", title = "short", message = "a much longer message") + ) + val summary = resource.buildOperatorExecutionSummary( + opId = "op-9", + state = "Failed", + resultMode = "table", + result = None, + totalTuplesCount = None, + consoleLogs = Some(logs) + ) + summary.errorMessages should have size 1 + summary.errorMessages.head.`type` shouldBe EXECUTION_FAILURE + summary.errorMessages.head.message shouldBe "a much longer message" + summary.errorMessages.head.operatorId shouldBe "op-9" + summary.consoleMessages.get should have size 2 + } + + it should "keep the ERROR title when it is longer than the message" in { + val logs = + List( + ConsoleMessageSummary(msgType = "ERROR", title = "a long descriptive title", message = "") + ) + val summary = resource.buildOperatorExecutionSummary( + opId = "op-2", + state = "Failed", + resultMode = "table", + result = None, + totalTuplesCount = None, + consoleLogs = Some(logs) + ) + summary.errorMessages.head.message shouldBe "a long descriptive title" + } + + it should "keep the ERROR title when the message is non-empty but shorter" in { + // Exercises `message.nonEmpty` true AND `message.length > title.length` false. + val logs = + List( + ConsoleMessageSummary( + msgType = "ERROR", + title = "a fairly long error title", + message = "short" + ) + ) + val summary = resource.buildOperatorExecutionSummary( + opId = "op-5", + state = "Failed", + resultMode = "table", + result = None, + totalTuplesCount = None, + consoleLogs = Some(logs) + ) + summary.errorMessages.head.message shouldBe "a fairly long error title" + } + + it should "leave the result summary empty when no result was materialized" in { + val summary = resource.buildOperatorExecutionSummary( + opId = "op-3", + state = "Uninitialized", + resultMode = "table", + result = None, + totalTuplesCount = None, + consoleLogs = None + ) + summary.resultSummary shouldBe None + summary.consoleMessages shouldBe None + summary.errorMessages shouldBe empty + } + + // --- collectOperatorInfos (private wrapper around state/result/console collection) ------- + + "collectOperatorInfos" should "return an empty map when there are no target or stats operators" in { + val stateStore = new ExecutionStateStore + val executionService = buildExecutionService(stateStore) + + val operatorInfos = resource invokePrivate collectOperatorInfos( + ExecutionIdentity(1L), + executionService, + List.empty[String], + 100000, + 100000, + None + ) + + operatorInfos shouldBe empty + } + + it should "summarize an explicit target even when stats are absent" in { + val stateStore = new ExecutionStateStore + val executionService = buildExecutionService(stateStore) + + val operatorInfos = resource invokePrivate collectOperatorInfos( + insertExecutionRow(), + executionService, + List("target-op"), + 100000, + 100000, + None + ) + + operatorInfos.keySet shouldBe Set("target-op") + operatorInfos("target-op").state shouldBe "Unknown" + operatorInfos("target-op").resultSummary shouldBe None + operatorInfos("target-op").consoleMessages shouldBe None + } + + it should "include in-memory console-error operators that are not target operators" in { + val stateStore = new ExecutionStateStore + val executionService = buildExecutionService(stateStore) + val consoleMessage = new ConsoleMessage( + "worker-1", + Timestamp(Instant.now), + ConsoleMessageType.ERROR, + "source", + "short", + "a longer in-memory console error" + ) + val consoleState = ConsoleMessageProcessor.addMessageToOperatorConsole( + new ExecutionConsoleStore(), + "console-op", + consoleMessage, + 10 + ) + + val operatorInfos = resource invokePrivate collectOperatorInfos( + insertExecutionRow(), + executionService, + List.empty[String], + 100000, + 100000, + Some(consoleState) + ) + + operatorInfos.keySet shouldBe Set("console-op") + operatorInfos("console-op").errorMessages should have size 1 + operatorInfos( + "console-op" + ).errorMessages.head.message shouldBe "a longer in-memory console error" + operatorInfos("console-op").consoleMessages.get.map(_.msgType) shouldBe List("ERROR") + } + + it should "map operator stats state when the stats store has target metrics" in { + val stateStore = new ExecutionStateStore + stateStore.statsStore.updateState(_ => + ExecutionStatsStore( + operatorInfo = Map( + "stats-op" -> OperatorMetrics( + operatorState = WorkflowAggregatedState.KILLED, + operatorStatistics = OperatorStatistics() + ) + ) + ) + ) + val executionService = buildExecutionService(stateStore) + + val operatorInfos = resource invokePrivate collectOperatorInfos( + insertExecutionRow(), + executionService, + List("stats-op"), + 100000, + 100000, + None + ) + + operatorInfos("stats-op").state shouldBe "Killed" + } + + // --- executeWorkflowSync public branches ------------------------------------------------ + + "executeWorkflowSync" should "return an error when init does not publish an execution service" in { + val summary = runWithStubWorkflow(_ => ()) + + summary.success shouldBe false + summary.state shouldBe "Error" + summary.operators shouldBe empty + summary.errors should have size 1 + summary.errors.head.`type` shouldBe EXECUTION_FAILURE + summary.errors.head.message shouldBe "Failed to initialize execution service" + } + + it should "return an error when waiting for execution state fails" in { + val stateStore = new ExecutionStateStore + val executionService = buildExecutionService(stateStore) + + val summary = runWithStubWorkflow( + service => { + service.executionService.onNext(executionService) + val failThread = new Thread(() => { + Thread.sleep(50) + failMetadataObservable(stateStore, new RuntimeException("metadata stream failed")) + }) + failThread.setDaemon(true) + failThread.start() + }, + syncRequest(timeoutSeconds = 1) + ) + + summary.success shouldBe false + summary.state shouldBe "Error" + summary.operators shouldBe empty + summary.errors should have size 1 + summary.errors.head.`type` shouldBe EXECUTION_FAILURE + summary.errors.head.message shouldBe "metadata stream failed" + } + + it should "assemble a completed summary when execution is already terminal" in { + val stateStore = new ExecutionStateStore + stateStore.metadataStore.updateState(_.withState(WorkflowAggregatedState.COMPLETED)) + val executionService = buildExecutionService(stateStore) + + val summary = runWithStubWorkflow(_.executionService.onNext(executionService)) + + summary.success shouldBe true + summary.state shouldBe "Completed" + summary.operators shouldBe empty + summary.errors shouldBe empty + } + + // --- assembleExecutionSummary (extracted from executeWorkflowSync) ---------------------- + + private def metadataStore( + state: WorkflowAggregatedState, + fatalErrors: Seq[WorkflowFatalError] = Seq.empty + ): ExecutionMetadataStore = + ExecutionMetadataStore( + state = state, + fatalErrors = fatalErrors, + executionId = ExecutionIdentity(0L) + ) + + private def failingOperatorSummary: OperatorExecutionSummary = + OperatorExecutionSummary( + state = "Failed", + errorMessages = + List(WorkflowFatalError(EXECUTION_FAILURE, Timestamp(Instant.now), "err", "", "op1")), + resultSummary = None, + consoleMessages = None + ) + + "assembleExecutionSummary" should "report success for a COMPLETED run with no errors" in { + val summary = resource.assembleExecutionSummary( + finalState = metadataStore(WorkflowAggregatedState.COMPLETED), + operatorInfos = Map.empty, + terminatedByConsoleError = false, + terminatedByTargetResults = false + ) + summary.success shouldBe true + summary.state shouldBe "Completed" + summary.errors shouldBe empty + summary.operators shouldBe empty + } + + it should "map a non-terminal-success final state through stateToString" in { + val summary = resource.assembleExecutionSummary( + finalState = metadataStore(WorkflowAggregatedState.FAILED), + operatorInfos = Map.empty, + terminatedByConsoleError = false, + terminatedByTargetResults = false + ) + summary.success shouldBe false + summary.state shouldBe "Failed" + } + + it should "force a Failed state when terminated by a console error, regardless of final state" in { + val summary = resource.assembleExecutionSummary( + finalState = metadataStore(WorkflowAggregatedState.COMPLETED), + operatorInfos = Map.empty, + terminatedByConsoleError = true, + terminatedByTargetResults = false + ) + summary.state shouldBe "Failed" + summary.success shouldBe false + } + + it should "override to Completed/success when terminated by target results on a non-completed state" in { + val summary = resource.assembleExecutionSummary( + finalState = metadataStore(WorkflowAggregatedState.RUNNING), + operatorInfos = Map.empty, + terminatedByConsoleError = false, + terminatedByTargetResults = true + ) + summary.state shouldBe "Completed" + summary.success shouldBe true + } + + it should "prefer console-error failure over target-results completion" in { + val summary = resource.assembleExecutionSummary( + finalState = metadataStore(WorkflowAggregatedState.RUNNING), + operatorInfos = Map.empty, + terminatedByConsoleError = true, + terminatedByTargetResults = true + ) + summary.state shouldBe "Failed" + summary.success shouldBe false + } + + it should "mark the run unsuccessful when any operator reports console errors" in { + val summary = resource.assembleExecutionSummary( + finalState = metadataStore(WorkflowAggregatedState.COMPLETED), + operatorInfos = Map("op1" -> failingOperatorSummary), + terminatedByConsoleError = false, + terminatedByTargetResults = false + ) + summary.success shouldBe false + summary.state shouldBe "Completed" + } + + it should "mark target-results completion unsuccessful when an operator reports console errors" in { + val summary = resource.assembleExecutionSummary( + finalState = metadataStore(WorkflowAggregatedState.RUNNING), + operatorInfos = Map("op1" -> failingOperatorSummary), + terminatedByConsoleError = false, + terminatedByTargetResults = true + ) + summary.success shouldBe false + summary.state shouldBe "Completed" + } + + it should "surface each final-state fatal error" in { + val summary = resource.assembleExecutionSummary( + finalState = metadataStore( + WorkflowAggregatedState.COMPLETED, + Seq(WorkflowFatalError(EXECUTION_FAILURE, Timestamp(Instant.now), "boom", "", "op1")) + ), + operatorInfos = Map.empty, + terminatedByConsoleError = false, + terminatedByTargetResults = false + ) + summary.errors should have size 1 + summary.errors.head.`type` shouldBe EXECUTION_FAILURE + summary.errors.head.message shouldBe "boom" + } +}