Yicong-Huang commented on code in PR #6009:
URL: https://github.com/apache/texera/pull/6009#discussion_r3564620220
##########
frontend/src/app/workspace/component/agent/agent-interaction/agent-interaction.component.ts:
##########
@@ -192,95 +191,34 @@ export class AgentInteractionComponent implements OnInit,
OnChanges {
}
/**
- * Get column names from sample records, placing __row_index__ first
(displayed as "Row").
+ * Get visible column names from sample rows.
Review Comment:
use tuple
##########
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> = {}): OperatorInfo {
+import {
+ ConsoleMessageType,
+ OperatorState,
+ WorkflowFatalErrorType,
+ type OperatorExecutionSummary,
+ type WebOutputMode,
+ 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<string, any>): 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<string, any>[]): [number, Tuple][] {
Review Comment:
rename with Tuple?
##########
agent-service/src/types/execution.ts:
##########
@@ -17,56 +17,121 @@
* under the License.
*/
-interface ConsoleMessage {
- msgType: string;
+export enum WorkflowFatalErrorType {
+ COMPILATION_ERROR = "COMPILATION_ERROR",
+ EXECUTION_FAILURE = "EXECUTION_FAILURE",
+}
+
+// Canonical agent-service error projection. It follows the engine's
+// workflowruntimestate.proto shape so compile and execution errors share one
model.
+// The legacy sync-execution adapter synthesizes metadata the backend does not
yet emit.
+// 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;
+}
+
+// A normalized result row using the engine Tuple shape: a schema plus
positional fields.
+// The legacy backend returns JSON records, so the adapter builds a synthetic
all-STRING
+// schema after the backend has truncated values for display.
+export interface Attribute {
+ attributeName: string;
+ attributeType: string;
+}
+
+export interface Schema {
+ attributes: Attribute[];
+}
+
+export interface Tuple {
+ schema: Schema;
+ fields: unknown[];
+}
Review Comment:
can you make all interfaces readonly? also readonly array etc.
##########
agent-service/src/api/execution-api.ts:
##########
@@ -17,6 +17,19 @@
* under the License.
*/
+import { z } from "zod";
Review Comment:
I am not familiar with zod, can you explain in code comment and here, why do
we need to use it to construct schema?
##########
frontend/src/app/workspace/component/agent/agent-interaction/agent-interaction.component.ts:
##########
@@ -192,95 +191,34 @@ export class AgentInteractionComponent implements OnInit,
OnChanges {
}
/**
- * Get column names from sample records, placing __row_index__ first
(displayed as "Row").
+ * Get visible column names from sample rows.
*/
public getSampleColumns(): string[] {
- if (!this.sampleRecords || this.sampleRecords.length === 0) return [];
- const allKeys = Object.keys(this.sampleRecords[0]);
- const rowIndexKey = allKeys.find(k => k.startsWith("_") &&
k.includes("row_index"));
- const otherKeys = allKeys.filter(k => k !== rowIndexKey);
- return rowIndexKey ? [rowIndexKey, ...otherKeys] : otherKeys;
+ if (!this.sampleTuples || this.sampleTuples.length === 0) return [];
+ return tupleColumns(this.sampleTuples[0][1]);
}
/**
* Get display name for a column header.
*/
public getColumnDisplayName(col: string): string {
- if (col.startsWith("_") && col.includes("row_index")) return "Row";
return col;
}
Review Comment:
this is NoOp?
##########
agent-service/src/api/execution-api.ts:
##########
@@ -36,3 +49,166 @@ export interface LogicalPlan {
opsToViewResult?: string[];
opsToReuseResult?: string[];
}
+
+const legacyConsoleMessageSchema = z.object({
+ msgType: z.nativeEnum(ConsoleMessageType),
+ title: z.string().default(""),
+ message: z.string(),
+});
+
+const legacyResultModeSchema = z.enum(["table", "visualization"]);
+type LegacyResultMode = z.output<typeof legacyResultModeSchema>;
+
+const legacyResultModeToWebOutputMode = {
+ table: { type: "PaginationMode" },
+ visualization: { type: "SetSnapshotMode" },
+} satisfies Record<LegacyResultMode, WebOutputMode>;
+
+const legacyOperatorInfoSchema = z
+ .object({
+ state: z.string(),
+ inputTuples: z.number(),
+ outputTuples: z.number(),
+ resultMode: legacyResultModeSchema,
+ result: z.array(z.record(z.unknown())).nullish(),
+ totalRowCount: z.number().int().nonnegative().nullish(),
+ displayedRows: z.number().int().nonnegative().nullish(),
+ truncated: z.boolean().nullish(),
+ consoleLogs: z.array(legacyConsoleMessageSchema).nullish(),
+ error: z.string().nullish(),
+ warnings: z.array(z.string()).nullish(),
+ })
+ .passthrough();
Review Comment:
why do we keep legacy definition, is that still needed?
##########
frontend/src/app/workspace/component/agent/agent-interaction/agent-interaction.component.ts:
##########
@@ -192,95 +191,34 @@ export class AgentInteractionComponent implements OnInit,
OnChanges {
}
/**
- * Get column names from sample records, placing __row_index__ first
(displayed as "Row").
+ * Get visible column names from sample rows.
*/
public getSampleColumns(): string[] {
- if (!this.sampleRecords || this.sampleRecords.length === 0) return [];
- const allKeys = Object.keys(this.sampleRecords[0]);
- const rowIndexKey = allKeys.find(k => k.startsWith("_") &&
k.includes("row_index"));
- const otherKeys = allKeys.filter(k => k !== rowIndexKey);
- return rowIndexKey ? [rowIndexKey, ...otherKeys] : otherKeys;
+ if (!this.sampleTuples || this.sampleTuples.length === 0) return [];
+ return tupleColumns(this.sampleTuples[0][1]);
}
/**
* Get display name for a column header.
*/
public getColumnDisplayName(col: string): string {
- if (col.startsWith("_") && col.includes("row_index")) return "Row";
return col;
}
- /**
- * Parse resultStatistics into displayable column stats.
- * Each entry in resultStatistics is a JSON string with { data_type,
statistics: { ... } }.
- */
- public getParsedColumnStats(): Array<{
- column: string;
- dataType: string;
- stats: Array<{ key: string; value: string }>;
- }> {
- if (!this.resultStatistics) return [];
- const sampleCols = this.getSampleColumns().filter(c => !c.startsWith("_")
|| !c.includes("row_index"));
- const columns = sampleCols.length > 0 ? sampleCols :
Object.keys(this.resultStatistics);
- const result: Array<{ column: string; dataType: string; stats: Array<{
key: string; value: string }> }> = [];
- const excludedKeys = new Set(["count", "std", "p25", "median", "p75"]);
-
- for (const colName of columns) {
- const statsJson = this.resultStatistics[colName];
- if (!statsJson) continue;
- try {
- const parsed = JSON.parse(statsJson);
- const dataType: string = parsed.data_type ?? "unknown";
- const statistics: Record<string, any> = parsed.statistics ?? {};
- const statEntries: Array<{ key: string; value: string }> = [];
-
- for (const [key, value] of Object.entries(statistics)) {
- if (value === undefined || excludedKeys.has(key)) continue;
- if (key === "top_10" && typeof value === "object") {
- const topEntries = Object.entries(value as Record<string, any>)
- .slice(0, 5)
- .map(([k, v]) => `${k}: ${v}`)
- .join(", ");
- statEntries.push({ key: "top values", value: topEntries });
- } else if (value === null || String(value) === "null") {
- statEntries.push({ key, value: "NaN" });
- } else if (typeof value !== "object") {
- const formatted =
- typeof value === "number" && !Number.isInteger(value)
- ? Number(value.toPrecision(4)).toString()
- : String(value);
- statEntries.push({ key, value: formatted });
- }
- }
- result.push({ column: colName, dataType, stats: statEntries });
- } catch {
- // skip unparseable
- }
- }
- return result;
- }
-
- public hasColumnStats(): boolean {
- return this.getParsedColumnStats().length > 0;
- }
-
- public getDisplayRows(): Array<{ record?: Record<string, any>; isEllipsis:
boolean }> {
- if (!this.sampleRecords || this.sampleRecords.length === 0) return [];
- const rowIndexKey = Object.keys(this.sampleRecords[0]).find(k =>
k.startsWith("_") && k.includes("row_index"));
- if (!rowIndexKey) {
- return this.sampleRecords.map(r => ({ record: r, isEllipsis: false }));
- }
+ public getDisplayRows(): Array<{ rowIndex?: number; row?: Record<string,
unknown>; isEllipsis: boolean }> {
Review Comment:
use Tuple semantic.
Why do we need to display tuples?
--
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]
