How to run triggers during a load or unload

A data-operation trigger runs custom logic inside the same datastore transaction as a data load or unload. It reads the rows an operation has just changed, even though they are not yet committed, and writes derived rows back into the same transaction. The motivating use case is maintaining a data-availability table, the distinct as-of dates present in a trades table, atomically with the operations on the trades, so the availability table stays consistent with the data even on a partial update or a removal.

Prerequisites

Triggers extend a working data load. Start from data loading, which covers topics, channels, and field calculators.

What is a data-operation trigger?

A topic fetches rows from an external source; a trigger does not. Instead, a trigger runs after its ordering predecessors and works against the in-flight transaction:

it reads the rows already changed in the current transaction (the rows a topic just loaded, or the rows an unload just removed), and
it writes derived rows into the same transaction.

Because everything happens in one transaction, the topic rows and the trigger’s derived rows become visible together on commit. If the trigger throws, the whole operation (topic rows included) is rolled back, so there is never a partial state. A trigger runs only when it fires: it declares the tables it reads, and it fires whenever a requested topic feeds one of those tables, on a load or an unload.

How to implement a trigger

Implement IDataOperationTrigger:

getInputTables(): the tables the trigger reads. The trigger fires whenever a requested topic feeds one of them, and runs after every topic and trigger that modified them, so it sees their complete content.
getOutputTables(): the tables the trigger writes to. The transaction is opened over these tables, in addition to the topics’ target tables, so the trigger’s writes have an open transaction.
execute(queryManager, transaction, context): the logic. Use queryManager to build the read query, transaction to query the in-flight view and to add or remove rows, and context for the operation scope() and operationType().

To stay consistent across load and unload, write the trigger as a rebuild: clear its derived rows for the scope, then re-derive them from the current input rows. On a load this replaces the scope’s rows; on an unload the input rows are gone, so the orphaned derived rows stay removed. The following trigger rebuilds a data-availability table: it clears the table, then queries the distinct as-of dates of the trades from the in-flight transaction and writes one row per date:

/** Keeps a data-availability table equal to the distinct as-of dates present in the trades. */
public record DataAvailabilityTrigger(String tradesTable, String availabilityTable)
    implements IDataOperationTrigger {
  @Override
  public Set<String> getInputTables() {
    // The trigger fires whenever a requested topic feeds this table, and runs after it.
    return Set.of(tradesTable);
  }
  @Override
  public Set<String> getOutputTables() {
    // The transaction is opened over every table the trigger writes to.
    return Set.of(availabilityTable);
  }
  @Override
  public void execute(
      final IQueryManager queryManager,
      final IOpenedTransaction transaction,
      final ITriggerContext context) {
    // Rebuild the availability table from scratch: clear it, then re-derive the distinct as-of
    // dates from the current trades. This keeps it correct for both operations: on a load the
    // dates of the added trades appear; on an unload (context.operationType() == UNLOAD) the
    // removed trades are no longer seen, so the dates they alone carried disappear.
    try {
      transaction.removeWhere(availabilityTable, BaseConditions.TRUE);
    } catch (DatastoreTransactionException e) {
      throw new RuntimeException(e);
    }
    final var query =
        queryManager
            .distinctQuery()
            .forTable(tradesTable)
            .withoutCondition()
            .withTableFields("AsOfDate")
            .toQuery();
    // The transaction's query runner sees the rows the operation just changed in this
    // not-yet-committed transaction, so the availability table and the trades commit as one
    // consistent state: "which dates hold data?" is never wrong, even on a partial update.
    try (var cursor = transaction.getQueryRunner().distinctQuery(query).runCurrentThread()) {
      while (cursor.next()) {
        transaction.add(availabilityTable, cursor.getRecord().read(0));
      }
    }
  }
}

How to register a trigger

Register triggers on configurator.triggers() with a unique name. A trigger is not wired to a topic explicitly: it fires and is ordered by the tables it reads (getInputTables()), so it runs whenever a requested topic feeds one of those tables:

@Bean
public IDataLoadConfigurer dataAvailabilityConfigurer() {
  return configurator -> {
    configurator.channels().addChannel("TradesChannel", "Trades");
    configurator
        .topics()
        .addTopic("tradesTopic", ILocalCsvTopicConfigurator.class)
        .path("/data/trades")
        .filePattern("glob:*.csv")
        .parsersFromTable("Trades")
        .attachChannel("TradesChannel");
    // The trigger reads Trades, so it fires whenever a requested topic feeds that table
    // (here, tradesTopic via TradesChannel). No explicit ordering is needed.
    configurator
        .triggers()
        .addTrigger(
            "dataAvailability", new DataAvailabilityTrigger("Trades", "DataAvailability"));
  };
}

How to order a topic after another topic

configurator.topics().orderTopicAfterTopic(topicName, afterTopicName) schedules a topic after another topic and all of the triggers fired by that topic. Use it when one topic must load only once another topic’s rows have been loaded and its triggers have run:

// positionsTopic loads only after tradesTopic AND the triggers it fires (such as
// dataAvailability) have completed in the same load.
configurator.topics().orderTopicAfterTopic("positionsTopic", "tradesTopic");

How an operation with triggers runs

Configuring a trigger does not change the default load when nothing is ordered: all requested topics still load together in a single batch. Otherwise, the requested topics and the triggers they fire run in this order:

all currently-unconstrained topics load together in one batch;
each fired trigger runs, in dependency order, reading the in-flight view and writing into the same transaction;
any topic ordered after another topic loads once that topic and its triggers have completed.

All of these steps run inside one datastore transaction, which is committed once at the end. Ordering references to topics or triggers that are not part of the current operation are ignored, and a cycle in the ordering relations fails the operation.

Triggers and unloading

Triggers run on an unload too. A DlcUnloadRequest removes the rows of its topics through their unload conditions and then, in the same transaction, runs every trigger fired by a table those topics feed, just as a load does. Because a rebuild trigger re-derives its target from the current input rows, the rows it had derived from the now-removed input rows are cleaned up automatically. This is why the rebuild pattern matters: a trigger that only adds derived rows keeps its derived table correct on load but leaves orphaned rows behind on unload. Clearing the derived rows for the scope before re-deriving keeps the table consistent for both operations. A trigger that must do something different on each can branch on context.operationType().

How to override a solution’s trigger

Trigger names must be unique. To override a pre-defined trigger, replace it under the same name with replaceTrigger; firing follows the replacement’s own input tables:

// Override a trigger defined by Solutions: replace it under the same name.
configurator
    .triggers()
    .replaceTrigger(
        "dataAvailability", new DataAvailabilityTrigger("Trades", "DataAvailability"));

Data loading: topics, channels, column renames, and field calculators.

​Prerequisites

​What is a data-operation trigger?

​How to implement a trigger

​How to register a trigger

​How to order a topic after another topic

​How an operation with triggers runs

​Triggers and unloading

​How to override a solution’s trigger

​Related pages