diff --git a/frontend/src/app/workspace/component/menu/menu.component.spec.ts b/frontend/src/app/workspace/component/menu/menu.component.spec.ts index 3a5f8440b80..b7e6884819c 100644 --- a/frontend/src/app/workspace/component/menu/menu.component.spec.ts +++ b/frontend/src/app/workspace/component/menu/menu.component.spec.ts @@ -752,7 +752,7 @@ describe("MenuComponent", () => { const convert = vi .spyOn(component["workflowToPythonService"], "convertToPython") .mockReturnValue(of({ type: "success", pythonCode: "print(1)" } as any)); - const validSubgraph = vi.spyOn(validationWorkflowService, "getValidTexeraGraph"); + const validSubgraph = vi.spyOn(validationWorkflowService, "getValidOperatorIDs"); workflowActionService.addOperator(mockScanPredicate, mockPoint); validationStream$.next({ errors: {}, workflowEmpty: false }); diff --git a/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts b/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts index 28f975540e1..d73589c3e6d 100644 --- a/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts +++ b/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts @@ -649,7 +649,7 @@ describe("WorkflowCompilingService compile pipeline", () => { workflowActionService.addOperator(mockScanPredicate, mockPoint); workflowActionService.addOperator(mockResultPredicate, mockPoint); workflowActionService.addLink(mockScanResultLink); - // ScanSource requires `tableName`; without it the operator is filtered out of the valid graph + // ScanSource requires `tableName`; without it the operator is filtered out of the logical plan workflowActionService.setOperatorProperty(mockScanPredicate.operatorID, { tableName: "twitter" }); tick(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS); return httpTestingController.expectOne(compileUrl); @@ -743,6 +743,56 @@ describe("WorkflowCompilingService compile pipeline", () => { expect(service.getWorkflowCompilationState()).toBe(CompilationState.Succeeded); warn.mockRestore(); })); + + it("leaves an invalid operator, and the links touching it, out of the logical plan", fakeAsync(() => { + workflowActionService.addOperator(mockScanPredicate, mockPoint); + workflowActionService.addOperator(mockResultPredicate, mockPoint); + workflowActionService.addLink(mockScanResultLink); + // no tableName, which ScanSource requires + tick(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS); + const request = httpTestingController.expectOne(compileUrl); + + const body = JSON.parse(request.request.body); + expect(body.operators.map((operator: any) => operator.operatorID)).toEqual([mockResultPredicate.operatorID]); + expect(body.links).toEqual([]); + + request.flush({ operatorOutputSchemas: {}, operatorErrors: {} }); + })); + + // Compiling used to plan a filtered copy of the graph: a new WorkflowGraph, whose SharedModel opened a + // y-websocket connection to a room of its own that nothing ever closed, one more on every debounced edit. + it("opens no websocket when it compiles", fakeAsync(() => { + const openedUrls: string[] = []; + vi.stubGlobal( + "WebSocket", + class { + constructor(url: string) { + openedUrls.push(url); + } + } + ); + try { + buildWorkflowAndCompile().flush({ operatorOutputSchemas: {}, operatorErrors: {} }); + workflowActionService.setOperatorProperty(mockScanPredicate.operatorID, { tableName: "reddit" }); + tick(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS); + httpTestingController.expectOne(compileUrl).flush({ operatorOutputSchemas: {}, operatorErrors: {} }); + } finally { + vi.unstubAllGlobals(); + } + + expect(openedUrls).toEqual([]); + })); + + it("stops compiling once the service is destroyed", fakeAsync(() => { + buildWorkflowAndCompile().flush({ operatorOutputSchemas: {}, operatorErrors: {} }); + + // what the injector calls on teardown; @UntilDestroy() adds it to the class + (service as any).ngOnDestroy(); + workflowActionService.setOperatorProperty(mockScanPredicate.operatorID, { tableName: "reddit" }); + tick(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS); + + httpTestingController.expectNone(compileUrl); + })); }); describe("WorkflowCompilingService.applySchemaPropagationResult without a propagated input schema", () => { diff --git a/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.ts b/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.ts index dae47a0088b..07820f2e5cf 100644 --- a/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.ts +++ b/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.ts @@ -28,6 +28,7 @@ import { ExecuteWorkflowService } from "../execute-workflow/execute-workflow.ser import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; import { catchError, debounceTime, mergeMap } from "rxjs/operators"; import { DynamicSchemaService } from "../dynamic-schema/dynamic-schema.service"; +import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; import { AttributeType, CompilationState, @@ -61,6 +62,7 @@ export const WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS = 500; * Specifically for schema autocomplete, by contract, property name `attribute` and `attributes` indicate the field is a column of the operator's input, * and schema propagation can provide autocomplete for the column names. */ +@UntilDestroy() @Injectable({ providedIn: "root", }) @@ -95,12 +97,19 @@ export class WorkflowCompilingService { .pipe(debounceTime(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS)) .pipe( mergeMap(() => { - const logicalPlan = ExecuteWorkflowService.getLogicalPlanRequest( - this.validationWorkflowService.getValidTexeraGraph(), - undefined - ); - return this.compile(logicalPlan); - }) + // Plan the live graph, then keep its valid operators and the links between them. Planning a filtered + // copy of the graph instead builds a new WorkflowGraph per compile, and its SharedModel opens a + // y-websocket connection that nothing ever closes. + const logicalPlan = ExecuteWorkflowService.getLogicalPlanRequest(this.workflowActionService.getTexeraGraph()); + const validOperatorIDs = this.validationWorkflowService.getValidOperatorIDs(); + return this.compile({ + operators: logicalPlan.operators.filter(operator => validOperatorIDs.has(operator.operatorID)), + links: logicalPlan.links.filter( + link => validOperatorIDs.has(link.fromOpId) && validOperatorIDs.has(link.toOpId) + ), + }); + }), + untilDestroyed(this) ) .subscribe(response => { if (response.physicalPlan) { @@ -299,7 +308,7 @@ export class WorkflowCompilingService { * that users can easily set the properties of the next operator. For eg: If there are two operators Source:Scan and KeywordSearch and * a link is created between them, the attributed of the table selected in Source can be propagated to the KeywordSearch operator. */ - private compile(logicalPlan: LogicalPlan): Observable { + private compile(logicalPlan: Pick): Observable { // create a Logical Plan based on the workflow graph // remove unnecessary information for schema propagation. const body = { diff --git a/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts b/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts index f0bee8f1c4e..f9dc285ef0c 100644 --- a/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts +++ b/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts @@ -555,30 +555,28 @@ describe("ValidationWorkflowService", () => { }); }); - describe("the graph sent to the compiler", () => { + describe("the operators sent to the compiler", () => { it("keeps an operator holding a reference inside a block, before any compile", () => { addBlockAround({ limit: "$K" }); - const validGraph = validationWorkflowService.getValidTexeraGraph(); - expect(validGraph.hasOperator(body.operatorID)).toBe(true); - expect(validGraph.getOperator(body.operatorID).operatorProperties).toEqual({ limit: "$K" }); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(true); // a name no enclosing Loop Start declares is no error either workflowActionservice.setOperatorProperty(body.operatorID, { limit: "$foo" }); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(true); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(true); }); it("leaves out one holding a reference outside every block, or with any other error", () => { addBlockAround({ limit: "$K", fraction: "warm" }); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(false); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(false); // a reference in an array property is an error everywhere workflowActionservice.setOperatorProperty(body.operatorID, { limit: 1, columns: "$K" }); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(false); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(false); // outside every block, a reference in an integer property is a type error workflowActionservice.setOperatorProperty(body.operatorID, { limit: "$K" }); workflowActionservice.deleteLinkWithID(mockScalaExecutorLoopEndLink.linkID); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(false); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(false); // and a valid operator stays in workflowActionservice.setOperatorProperty(body.operatorID, { limit: 1 }); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(true); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(true); }); }); }); diff --git a/frontend/src/app/workspace/service/validation/validation-workflow.service.ts b/frontend/src/app/workspace/service/validation/validation-workflow.service.ts index fb586513ba1..95c0c47b9e1 100644 --- a/frontend/src/app/workspace/service/validation/validation-workflow.service.ts +++ b/frontend/src/app/workspace/service/validation/validation-workflow.service.ts @@ -26,7 +26,6 @@ import Ajv, { ErrorObject } from "ajv"; import { map } from "rxjs/operators"; import { DynamicSchemaService } from "../dynamic-schema/dynamic-schema.service"; import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; -import { WorkflowGraph, WorkflowGraphReadonly } from "../workflow-graph/model/workflow-graph"; import { collectReferences, isReference, @@ -414,28 +413,17 @@ export class ValidationWorkflowService { } /** - * Gets a filtered version of the TexeraGraph containing only valid operators and their corresponding links. - * This method will create a copy of the TexeraGraph and do the validation on top of it. + * Gets the IDs of the operators in the TexeraGraph that pass validation, validating each one afresh. + * The compiler is sent these operators and the links between them. * - * @returns A json-schema-wise valid TexeraGraph + * @returns the IDs of the json-schema-wise valid operators */ - public getValidTexeraGraph(): WorkflowGraphReadonly { - const texeraGraph = this.workflowActionService.getTexeraGraph(); - const allOperators = texeraGraph.getAllOperators(); - const allLinks = texeraGraph.getAllLinks(); - - // Filter valid operators using validation service - const validOperators = allOperators.filter(operator => { - const validation = this.validateOperator(operator.operatorID); - return validation.isValid; - }); - - // Filter links to only include those connecting valid operators - const validOperatorIds = new Set(validOperators.map(op => op.operatorID)); - const validLinks = allLinks.filter( - link => validOperatorIds.has(link.source.operatorID) && validOperatorIds.has(link.target.operatorID) + public getValidOperatorIDs(): ReadonlySet { + return new Set( + this.workflowActionService + .getTexeraGraph() + .getAllOperatorIDs() + .filter(operatorID => this.validateOperator(operatorID).isValid) ); - - return new WorkflowGraph(validOperators, validLinks, texeraGraph.getAllCommentBoxes()); } }