Yicong-Huang commented on code in PR #6009:
URL: https://github.com/apache/texera/pull/6009#discussion_r3525255578
##########
amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala:
##########
@@ -73,38 +81,48 @@ 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
+// One sampled output row: the original row position plus the row's columns as
a
+// processed/truncated JSON object (not a raw engine Tuple, which would
serialize
+// as {schema, fields[]} and bypass the type-aware conversion + cell
truncation).
+// The index is carried explicitly rather than embedded in the tuple.
+@Generated
+case class SampleRow(
+ rowIndex: Int,
+ tuple: ObjectNode
Review Comment:
do we have a definition for Tuple?
##########
amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala:
##########
@@ -343,6 +344,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)
Review Comment:
the semantic of error in console message is "runtime exception", which are
fixable. shall we differentiate them from fatal errors (e.g., compilation
error) that cannot be fixed?
##########
amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala:
##########
@@ -73,38 +81,48 @@ 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
+// One sampled output row: the original row position plus the row's columns as
a
+// processed/truncated JSON object (not a raw engine Tuple, which would
serialize
+// as {schema, fields[]} and bypass the type-aware conversion + cell
truncation).
+// The index is carried explicitly rather than embedded in the tuple.
+@Generated
+case class SampleRow(
+ rowIndex: Int,
+ tuple: ObjectNode
)
-case class OperatorInfo(
- state: String,
- inputTuples: Long,
- outputTuples: Long,
- inputPortShapes: Option[List[PortShape]],
+@Generated
+case class OperatorResultSummary(
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]]
+ sampleTuples: List[SampleRow],
Review Comment:
let's not rename as `SampleRow`, keep Tuple. you can use
`List[typing.Tuple[Int, Tuple]]`
##########
agent-service/src/types/execution.ts:
##########
@@ -17,56 +17,116 @@
* 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<string, any>[];
- totalRowCount?: number;
- displayedRows?: number;
- truncated?: boolean;
- consoleLogs?: ConsoleMessage[];
- error?: string;
- warnings?: string[];
- resultStatistics?: Record<string, string>;
+// 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<string, OperatorInfo>;
- compilationErrors?: Record<string, string>;
- 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;
+}
+
+// One sampled output row: its original position plus the row's columns.
+export interface SampleRow {
+ rowIndex: number;
+ tuple: Record<string, unknown>;
+}
+
+export enum OperatorResultMode {
+ TABLE = "table",
+ VISUALIZATION = "visualization",
+}
Review Comment:
is this the same as backend definition? I recall we have more complex result
mode
##########
agent-service/src/types/execution.ts:
##########
@@ -17,56 +17,116 @@
* 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<string, any>[];
- totalRowCount?: number;
- displayedRows?: number;
- truncated?: boolean;
- consoleLogs?: ConsoleMessage[];
- error?: string;
- warnings?: string[];
- resultStatistics?: Record<string, string>;
+// 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<string, OperatorInfo>;
- compilationErrors?: Record<string, string>;
- 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;
+}
+
+// One sampled output row: its original position plus the row's columns.
+export interface SampleRow {
+ rowIndex: number;
+ tuple: Record<string, unknown>;
Review Comment:
need definition for Tuple.
##########
agent-service/src/agent/tools/result-formatting.spec.ts:
##########
@@ -19,104 +19,120 @@
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";
+import {
+ ConsoleMessageType,
+ OperatorState,
+ OperatorResultMode,
+ WorkflowFatalErrorType,
+ type OperatorExecutionSummary,
+ type WorkflowFatalError,
+ type SampleRow,
+} from "../../types/execution";
+
+function toSampleRows(rows: Record<string, any>[]): SampleRow[] {
+ return rows.map((tuple, rowIndex) => ({ rowIndex, tuple }));
+}
-function makeOpInfo(overrides: Partial<OperatorInfo> = {}): OperatorInfo {
- return {
- state: "completed",
- inputTuples: 0,
- outputTuples: 0,
- resultMode: "table",
- ...overrides,
- };
+interface OpInfoOverrides {
+ state?: OperatorState;
+ error?: string;
+ outputTuples?: number;
+ tuplesCount?: number;
+ warnings?: string[];
+ result?: Record<string, any>[];
+ sampleTuples?: SampleRow[];
Review Comment:
I feel better to use `Record<number, Tuple>[]`
##########
agent-service/src/types/execution.ts:
##########
@@ -17,56 +17,116 @@
* 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<string, any>[];
- totalRowCount?: number;
- displayedRows?: number;
- truncated?: boolean;
- consoleLogs?: ConsoleMessage[];
- error?: string;
- warnings?: string[];
- resultStatistics?: Record<string, string>;
+// 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<string, OperatorInfo>;
- compilationErrors?: Record<string, string>;
- 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;
+}
+
+// One sampled output row: its original position plus the row's columns.
+export interface SampleRow {
+ rowIndex: number;
+ tuple: Record<string, unknown>;
+}
+
+export enum OperatorResultMode {
+ TABLE = "table",
+ VISUALIZATION = "visualization",
+}
+
+// An operator's output, summarized for the agent. `sampleTuples` are the
+// symmetrically-truncated output rows (the middle is dropped, so `rowIndex`
+// values may have gaps). `outputSchema` / per-column statistics are intended
+// future additions — the engine does not produce them yet.
export interface OperatorResultSummary {
- state: string;
- inputTuples: number;
- outputTuples: number;
- inputPortShapes?: PortShape[];
- outputColumns?: number;
- error?: string;
- warnings?: string[];
- consoleLogCount?: number;
- totalRowCount?: number;
- sampleRecords?: Record<string, unknown>[];
- resultStatistics?: Record<string, string>;
+ resultMode: OperatorResultMode;
+ sampleTuples: SampleRow[];
+ // Total output rows before truncation (sampleTuples may hold fewer).
+ tuplesCount: 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<WorkflowFatalError>;
+ // 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;
Review Comment:
do we need this `success: boolean;`? It is a boolean info, and
`WorkflowExecutionState` should capture this info already.
##########
agent-service/src/types/execution.ts:
##########
@@ -17,56 +17,116 @@
* 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<string, any>[];
- totalRowCount?: number;
- displayedRows?: number;
- truncated?: boolean;
- consoleLogs?: ConsoleMessage[];
- error?: string;
- warnings?: string[];
- resultStatistics?: Record<string, string>;
+// 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<string, OperatorInfo>;
- compilationErrors?: Record<string, string>;
- 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;
+}
+
+// One sampled output row: its original position plus the row's columns.
+export interface SampleRow {
+ rowIndex: number;
+ tuple: Record<string, unknown>;
+}
+
+export enum OperatorResultMode {
+ TABLE = "table",
+ VISUALIZATION = "visualization",
+}
+
+// An operator's output, summarized for the agent. `sampleTuples` are the
+// symmetrically-truncated output rows (the middle is dropped, so `rowIndex`
+// values may have gaps). `outputSchema` / per-column statistics are intended
+// future additions — the engine does not produce them yet.
export interface OperatorResultSummary {
- state: string;
- inputTuples: number;
- outputTuples: number;
- inputPortShapes?: PortShape[];
- outputColumns?: number;
- error?: string;
- warnings?: string[];
- consoleLogCount?: number;
- totalRowCount?: number;
- sampleRecords?: Record<string, unknown>[];
- resultStatistics?: Record<string, string>;
+ resultMode: OperatorResultMode;
+ sampleTuples: SampleRow[];
Review Comment:
keep this the same, I still prefer you get rid of `Row` semantic and just
use `Tuple` for definition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
