<commit_before>import EventEmitter from "events";
import Long from "long";

import * as dbadapters from "df/api/dbadapters";
import { retry } from "df/api/utils/retry";
import { hashExecutionAction } from "df/api/utils/run_cache";
import { timingSafeEqual } from "df/common/strings";
import {
  JSONObjectStringifier,
  StringifiedMap,
  StringifiedSet
} from "df/common/strings/stringifier";
import { dataform } from "df/protos/ts";

const CANCEL_EVENT = "jobCancel";

const isSuccessfulAction = (actionResult: dataform.IActionResult) =>
  actionResult.status === dataform.ActionResult.ExecutionStatus.SUCCESSFUL ||
  actionResult.status === dataform.ActionResult.ExecutionStatus.CACHE_SKIPPED ||
  actionResult.status === dataform.ActionResult.ExecutionStatus.DISABLED;

export function run(
  dbadapter: dbadapters.IDbAdapter,
  graph: dataform.IExecutionGraph,
  partiallyExecutedRunResult?: dataform.IRunResult
): Runner {
  return new Runner(dbadapter, graph, partiallyExecutedRunResult).execute();
}

export class Runner {
  private readonly warehouseStateBeforeRunByTarget: StringifiedMap<
    dataform.ITarget,
    dataform.ITableMetadata
  >;
  private readonly warehouseStateAfterRunByTarget: StringifiedMap<
    dataform.ITarget,
    dataform.ITableMetadata
  >;
  private readonly persistedStateByTarget: StringifiedMap<
    dataform.ITarget,
    dataform.IPersistedTableMetadata
  >;
  private readonly nonTableDeclarationTargets: StringifiedSet<dataform.ITarget>;

  private readonly runResult: dataform.IRunResult;
  private readonly changeListeners: Array<(graph: dataform.IRunResult) => void> = [];
  private readonly eEmitter: EventEmitter;

  private pendingActions: dataform.IExecutionAction[];
  private stopped = false;
  private cancelled = false;
  private timeout: NodeJS.Timer;
  private timedOut = false;
  private executionTask: Promise<dataform.IRunResult>;

  private metadataReadPromises: Array<Promise<void>> = [];

  constructor(
    private readonly dbadapter: dbadapters.IDbAdapter,
    private readonly graph: dataform.IExecutionGraph,
    partiallyExecutedRunResult?: dataform.IRunResult
  ) {
    this.runResult = partiallyExecutedRunResult || {
      actions: []
    };
    this.warehouseStateBeforeRunByTarget = new StringifiedMap(
      JSONObjectStringifier.create(),
      graph.warehouseState.tables?.map(tableMetadata => [tableMetadata.target, tableMetadata])
    );
    this.warehouseStateAfterRunByTarget = new StringifiedMap(
      JSONObjectStringifier.create(),
      Array.from(this.warehouseStateBeforeRunByTarget.entries())
    );
    this.persistedStateByTarget = new StringifiedMap(
      JSONObjectStringifier.create(),
      graph.warehouseState.cachedStates?.map(persistedTableMetadata => [
        persistedTableMetadata.target,
        persistedTableMetadata
      ])
    );
    this.nonTableDeclarationTargets = new StringifiedSet<dataform.ITarget>(
      JSONObjectStringifier.create(),
      graph.declarationTargets.filter(
        declarationTarget =>
          this.warehouseStateBeforeRunByTarget.get(declarationTarget)?.type !==
          dataform.TableMetadata.Type.TABLE
      )
    );

    const completedActionNames = new Set(
      partiallyExecutedRunResult?.actions
        .filter(action => action.status !== dataform.ActionResult.ExecutionStatus.RUNNING)
        .map(action => action.name)
    );
    this.pendingActions = graph.actions.filter(action => !completedActionNames.has(action.name));
    this.eEmitter = new EventEmitter();
    // There could feasibly be thousands of listeners to this, 0 makes the limit infinite.
    this.eEmitter.setMaxListeners(0);
  }

  public onChange(listener: (graph: dataform.IRunResult) => Promise<void> | void): Runner {
    this.changeListeners.push(listener);
    return this;
  }

  public execute(): this {
    if (!!this.executionTask) {
      throw new Error("Executor already started.");
    }
    this.executionTask = this.executeGraph();
    if (!!this.graph.runConfig && !!this.graph.runConfig.timeoutMillis) {
      this.timeout = setTimeout(() => {
        this.timedOut = true;
        this.cancel();
      }, this.graph.runConfig.timeoutMillis);
    }
    return this;
  }<commit_msg>Fix run resumption for RunResults with zero actions. (#911)<commit_after>import EventEmitter from "events";
import Long from "long";

import * as dbadapters from "df/api/dbadapters";
import { retry } from "df/api/utils/retry";
import { hashExecutionAction } from "df/api/utils/run_cache";
import { timingSafeEqual } from "df/common/strings";
import {
  JSONObjectStringifier,
  StringifiedMap,
  StringifiedSet
} from "df/common/strings/stringifier";
import { dataform } from "df/protos/ts";

const CANCEL_EVENT = "jobCancel";

const isSuccessfulAction = (actionResult: dataform.IActionResult) =>
  actionResult.status === dataform.ActionResult.ExecutionStatus.SUCCESSFUL ||
  actionResult.status === dataform.ActionResult.ExecutionStatus.CACHE_SKIPPED ||
  actionResult.status === dataform.ActionResult.ExecutionStatus.DISABLED;

export function run(
  dbadapter: dbadapters.IDbAdapter,
  graph: dataform.IExecutionGraph,
  partiallyExecutedRunResult?: dataform.IRunResult
): Runner {
  return new Runner(dbadapter, graph, partiallyExecutedRunResult).execute();
}

export class Runner {
  private readonly warehouseStateBeforeRunByTarget: StringifiedMap<
    dataform.ITarget,
    dataform.ITableMetadata
  >;
  private readonly warehouseStateAfterRunByTarget: StringifiedMap<
    dataform.ITarget,
    dataform.ITableMetadata
  >;
  private readonly persistedStateByTarget: StringifiedMap<
    dataform.ITarget,
    dataform.IPersistedTableMetadata
  >;
  private readonly nonTableDeclarationTargets: StringifiedSet<dataform.ITarget>;

  private readonly runResult: dataform.IRunResult;
  private readonly changeListeners: Array<(graph: dataform.IRunResult) => void> = [];
  private readonly eEmitter: EventEmitter;

  private pendingActions: dataform.IExecutionAction[];
  private stopped = false;
  private cancelled = false;
  private timeout: NodeJS.Timer;
  private timedOut = false;
  private executionTask: Promise<dataform.IRunResult>;

  private metadataReadPromises: Array<Promise<void>> = [];

  constructor(
    private readonly dbadapter: dbadapters.IDbAdapter,
    private readonly graph: dataform.IExecutionGraph,
    partiallyExecutedRunResult?: dataform.IRunResult
  ) {
    this.runResult = {
      actions: [],
      ...partiallyExecutedRunResult
    };
    this.warehouseStateBeforeRunByTarget = new StringifiedMap(
      JSONObjectStringifier.create(),
      graph.warehouseState.tables?.map(tableMetadata => [tableMetadata.target, tableMetadata])
    );
    this.warehouseStateAfterRunByTarget = new StringifiedMap(
      JSONObjectStringifier.create(),
      Array.from(this.warehouseStateBeforeRunByTarget.entries())
    );
    this.persistedStateByTarget = new StringifiedMap(
      JSONObjectStringifier.create(),
      graph.warehouseState.cachedStates?.map(persistedTableMetadata => [
        persistedTableMetadata.target,
        persistedTableMetadata
      ])
    );
    this.nonTableDeclarationTargets = new StringifiedSet<dataform.ITarget>(
      JSONObjectStringifier.create(),
      graph.declarationTargets.filter(
        declarationTarget =>
          this.warehouseStateBeforeRunByTarget.get(declarationTarget)?.type !==
          dataform.TableMetadata.Type.TABLE
      )
    );

    const completedActionNames = new Set(
      this.runResult.actions
        .filter(action => action.status !== dataform.ActionResult.ExecutionStatus.RUNNING)
        .map(action => action.name)
    );
    this.pendingActions = graph.actions.filter(action => !completedActionNames.has(action.name));
    this.eEmitter = new EventEmitter();
    // There could feasibly be thousands of listeners to this, 0 makes the limit infinite.
    this.eEmitter.setMaxListeners(0);
  }

  public onChange(listener: (graph: dataform.IRunResult) => Promise<void> | void): Runner {
    this.changeListeners.push(listener);
    return this;
  }

  public execute(): this {
    if (!!this.executionTask) {
      throw new Error("Executor already started.");
    }
    this.executionTask = this.executeGraph();
    if (!!this.graph.runConfig && !!this.graph.runConfig.timeoutMillis) {
      this.timeout = setTimeout(() => {
        this.timedOut = true;
        this.cancel();
      }, this.graph.runConfig.timeoutMillis);
    }
    return this;
  }
{'fraction_non_alphanumeric': 0.10290299740382346, 'fraction_numerical': 0.0008260561718196837, 'mean_word_length': 4.228254164096237, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6157456', 'n_tokens_mistral': 2400, 'n_tokens_neox': 2392, 'n_words': 464}