Skip to content

Refactor Runner parameter overloads and resolve review comments #2210

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rafal-hawrylak wants to merge 1 commit into
base: main
Choose a base branch
from
+113 −17
Open

Refactor Runner parameter overloads and resolve review comments #2210

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 113 additions & 17 deletions cli/api/commands/run.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ export interface IExecutedAction {
}

export interface IExecutionOptions {
projectDir?: string;

@apilaskowski apilaskowski Jul 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this projectDir used downstream in latter PRs?

@apilaskowski apilaskowski Jul 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looking at this PR I can see that we are adding this field here, and then we are setting it (under some conditions to "."), but I am not sure how it is used. Possibly it is needed latter.

I can guess what this field does in general by the name, but is it needed?

@rafal-hawrylak rafal-hawrylak Jul 6, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, this is needed in the next PR - #2211. It is passed as the second arg to jitCompiler (via JitCompileChildProcess.compile) — it's the resolution root the worker needs to locate the caller's @dataform/core install. Plumbing it in here is the way to reduce the size of the next PR and still keep the reasonable amount of PRs.

@rafal-hawrylak rafal-hawrylak Jul 6, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, this is needed in the next PR — #2211. It is passed as the second arg to jitCompiler (via JitCompileChildProcess.compile) — it is the resolution root the worker needs to locate the caller's @dataform/core install. Plumbing it in here is the way to reduce the size of the next PR and still keep the reasonable amount of PRs.

bigquery?: {
jobPrefix?: string;
actionRetryLimit?: number;
Expand All @@ -32,23 +33,77 @@ export interface IExecutionOptions {
};
}

export type RunOptionsOrResult = IExecutionOptions | dataform.IRunResult;
export type CancelReason = "timeout" | "user" | "cancellation";

export function run(
dbadapter: dbadapters.IDbAdapter,
graph: dataform.IExecutionGraph,
executionOptions?: IExecutionOptions,
optionsOrResult: RunOptionsOrResult = {},
partiallyExecutedRunResult: dataform.IRunResult = {},
runnerNotificationPeriodMillis: number = flags.runnerNotificationPeriodMillis.get()
): Runner {
const { options, runResult } = handleParamsOverloads(optionsOrResult, partiallyExecutedRunResult);
return new Runner(
dbadapter,
graph,
executionOptions,
partiallyExecutedRunResult,
options,
runResult,
runnerNotificationPeriodMillis
).execute();
}

// Legacy `run()` and the positional-arg `Runner` constructor accept either
// `IExecutionOptions` or a partial `dataform.IRunResult` in the same slot.
// Discriminate on `.actions` presence (runResult has it, options doesn't).
// New callers should prefer `Runner.create` / `Runner.resume`, which pass
// unambiguous arguments through this helper.
function handleParamsOverloads(

@apilaskowski apilaskowski Jul 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not obvious to me why we are handling this that way. Can we put some information on what this function is supposed to do?

@rafal-hawrylak rafal-hawrylak Jul 6, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adding a doc block

optionsOrResult: RunOptionsOrResult,
partiallyExecutedRunResult: dataform.IRunResult
): { options: IExecutionOptions; runResult: dataform.IRunResult } {
if (optionsOrResult && (optionsOrResult as dataform.IRunResult).actions !== undefined) {

@apilaskowski apilaskowski Jul 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If an IRunResult object is passed with actions omitted or undefined, it will inadvertently be misclassified as IExecutionOptions, right? Is that expected?

return {
runResult: {
...(optionsOrResult as dataform.IRunResult)
},
options: { projectDir: "." }
};
}
const options: IExecutionOptions = {
projectDir: ".",
...(optionsOrResult as IExecutionOptions)
};
return {
options,
runResult: {
actions: [],
...partiallyExecutedRunResult
}
};
}

export class Runner {
public static create(
dbadapter: dbadapters.IDbAdapter,
graph: dataform.IExecutionGraph,
options: IExecutionOptions = {},
runnerNotificationPeriodMillis: number = flags.runnerNotificationPeriodMillis.get()
): Runner {
const { options: parsedOptions, runResult } = handleParamsOverloads(options, {});
return new Runner(dbadapter, graph, parsedOptions, runResult, runnerNotificationPeriodMillis);
}

public static resume(
dbadapter: dbadapters.IDbAdapter,
graph: dataform.IExecutionGraph,
runResult: dataform.IRunResult,
runnerNotificationPeriodMillis: number = flags.runnerNotificationPeriodMillis.get()
): Runner {
const { options, runResult: parsedRunResult } = handleParamsOverloads(runResult, {});
return new Runner(dbadapter, graph, options, parsedRunResult, runnerNotificationPeriodMillis);
}

private readonly warehouseStateByTarget: Map<string, dataform.ITableMetadata>;

private readonly allActionTargets: Set<string>;
Expand All @@ -63,22 +118,40 @@ export class Runner {
private cancelled = false;
private timeout: NodeJS.Timer;
private timedOut = false;
private skipReason: string = "";
private executionTask: Promise<dataform.IRunResult>;
private readonly executionOptions: IExecutionOptions;

constructor(

@apilaskowski apilaskowski Jul 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since we want to use create and resume from now on to increase readability, what about making those constructors private (or protected) to enforce usage of those methods? wdyt?

dbadapter: dbadapters.IDbAdapter,
graph: dataform.IExecutionGraph,
options: IExecutionOptions,
runResult: dataform.IRunResult,
runnerNotificationPeriodMillis?: number
);
constructor(
dbadapter: dbadapters.IDbAdapter,
graph: dataform.IExecutionGraph,
optionsOrResult?: RunOptionsOrResult,
partiallyExecutedRunResult?: dataform.IRunResult,
runnerNotificationPeriodMillis?: number
);
constructor(
private readonly dbadapter: dbadapters.IDbAdapter,
private readonly graph: dataform.IExecutionGraph,
private readonly executionOptions: IExecutionOptions = {},
optionsOrResult: RunOptionsOrResult = {},
partiallyExecutedRunResult: dataform.IRunResult = {},
private readonly runnerNotificationPeriodMillis: number = flags.runnerNotificationPeriodMillis.get()
) {
const { options, runResult } = handleParamsOverloads(
optionsOrResult,
partiallyExecutedRunResult
);
this.executionOptions = options;
this.runResult = runResult;
this.allActionTargets = new Set<string>(
graph.actions.map(action => targetStringifier.stringify(action.target))
);
this.runResult = {
actions: [],
...partiallyExecutedRunResult
};
this.warehouseStateByTarget = new Map<string, dataform.ITableMetadata>();
graph.warehouseState.tables?.forEach(tableMetadata =>
this.warehouseStateByTarget.set(
Expand Down Expand Up @@ -121,7 +194,7 @@ export class Runner {
const timeoutMillis = this.graph.runConfig.timeoutMillis - elapsedTimeMillis;
this.timeout = setTimeout(() => {
this.timedOut = true;
this.cancel();
this.cancel("timeout");
}, timeoutMillis);
}
return this;
Expand All @@ -131,7 +204,16 @@ export class Runner {
this.stopped = true;
}

public cancel() {
public cancel(reason: CancelReason = "user") {

@apilaskowski apilaskowski Jul 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would suggest preparing explicit map of possible reasons to avoid mistakes, and to make it more readable.

if (!this.skipReason) {
if (reason === "timeout") {
this.skipReason = `Run timed out after ${this.graph.runConfig.timeoutMillis / 1000} seconds`;

@apilaskowski apilaskowski Jul 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it possible that graph.runConfig.timeoutMillis is not defined? I am wondering if this might cause NaN to be displayed.

} else if (reason === "user") {
this.skipReason = "Run cancelled";
} else {
this.skipReason = "Run cancelled before action started";
}
}
this.cancelled = true;
this.eEmitter.emit(CANCEL_EVENT, undefined, undefined);
}
Expand Down Expand Up @@ -228,15 +310,28 @@ export class Runner {
if (this.cancelled) {
const allPendingActions = this.pendingActions;
this.pendingActions = [];
allPendingActions.forEach(pendingAction =>
const skipErrorMessage = this.skipReason || "Run cancelled before action started";
allPendingActions.forEach(pendingAction => {
let tasks: dataform.ITaskResult[];
if (pendingAction.tasks && pendingAction.tasks.length > 0) {
tasks = pendingAction.tasks.map((_, i) => ({
status: dataform.TaskResult.ExecutionStatus.SKIPPED,
...(i === 0 ? { errorMessage: skipErrorMessage } : {})
}));
} else {
tasks = [
{
status: dataform.TaskResult.ExecutionStatus.SKIPPED,
errorMessage: skipErrorMessage
}
];
}
this.runResult.actions.push({
target: pendingAction.target,
status: dataform.ActionResult.ExecutionStatus.SKIPPED,
tasks: pendingAction.tasks.map(() => ({
status: dataform.TaskResult.ExecutionStatus.SKIPPED
}))
})
);
tasks
});
});
this.notifyListeners();
return;
}
Expand Down Expand Up @@ -361,7 +456,8 @@ export class Runner {
}
} else {
actionResult.tasks.push({
status: dataform.TaskResult.ExecutionStatus.SKIPPED
status: dataform.TaskResult.ExecutionStatus.SKIPPED,
errorMessage: this.skipReason || "Task skipped after cancellation"
});
}
}
Expand Down
Loading