How to load tuples from a supplier

A tuple topic loads rows produced by a supplier function, rather than rows read from a source format such as CSV files or a JDBC query. The supplier builds each row in code, so the data can come from anywhere the supplier can reach: values computed on the fly, the parameters of the load request, a query against the datastore, or an external service such as a REST endpoint. The topic is configured through ITupleTopicConfigurator; the supplier and the names of the produced columns are described by ITupleTopicSettings. For the general topic and channel concepts, see the data loading overview.

How to load a fixed set of tuples

Call configurator.channels().addChannel(channelName, tableName) to declare the channel that loads rows into the target table.
Call configurator.topics().addTopic(topicName, ITupleTopicConfigurator.class) to declare a tuple topic.
Call settings(ITupleTopicSettings.of(supplier, columns)). The supplier returns the rows to load, and columns names the fields each row carries, in order.
Call attachChannel(channelName) on the topic to feed the channel.

Each row is an Object[] whose entries line up with the declared column names, so the first entry fills the first column, and so on. The following example loads two rows into the Sales table:

@Bean
public IDataLoadConfigurer loadTuplesConfigurer() {
  return configurator -> {
    configurator.channels().addChannel("SalesChannel", "Sales");
    configurator
        .topics()
        .addTopic("sales-tuples", ITupleTopicConfigurator.class)
        // The supplier returns the rows to load; each Object[] is one row whose fields
        // follow the order of the declared column names.
        .settings(
            ITupleTopicSettings.of(
                _ -> List.of(new Object[] {"trade0", 10}, new Object[] {"trade1", 11}),
                List.of("tradeId", "productId")))
        .attachChannel("SalesChannel");
  };
}

How the tuple supplier uses the load scope

The supplier is a Function<IDlcScope, Collection<Object[]>>. It receives the scope of the load request (the parameters supplied when the load is requested), so the produced rows can differ from one load to the next. The scope is a map of named parameters; read a parameter with a default when it may be absent.

configurator
    .topics()
    .addTopic("sales-tuples", ITupleTopicConfigurator.class)
    .settings(
        ITupleTopicSettings.of(
            // The scope carries the parameters of the load request, so the produced
            // rows can depend on it.
            scope -> {
              final int start = scope.<Integer>getOrDefault("startProductId", 0);
              return List.of(
                  new Object[] {"trade0", start}, new Object[] {"trade1", start + 1});
            },
            List.of("tradeId", "productId")))
    .attachChannel("SalesChannel");

How to build tuples from a datastore query

The supplier can read data already in the datastore and turn it into tuples. This is how a tuple topic derives a table from another table: one topic populates the source table, and the tuple topic queries it and loads a derived table. The same approach applies to any external source the supplier can call, such as a REST service. The supplier reads the datastore through the current transaction’s query runner (datastore.getTransactionManager().getQueryRunner()), so it sees the rows the ongoing load has already loaded, even before they are committed. For those rows to be there, the topic that feeds the queried table must run first: order the tuple topic after it with configurator.topics().orderTopicAfterTopic(topicName, afterTopicName). Both arguments are topic names, and both topics must already be registered. The following example loads a Products reference table from CSV files, then derives a ProductSummary table by querying Products and emitting one tuple per product:

@Bean
public IDataLoadConfigurer productSummaryConfigurer(final IDatastore datastore) {
  return configurator -> {
    configurator.channels().addChannel("ProductsChannel", "Products");
    configurator.channels().addChannel("ProductSummaryChannel", "ProductSummary");
    // A CSV topic populates the Products table from files on disk.
    configurator
        .topics()
        .addTopic("products-csv", ILocalCsvTopicConfigurator.class)
        .path("/data/products")
        .filePattern("glob:*.csv")
        .parsersFromTable("Products")
        .attachChannel("ProductsChannel");
    // A tuple topic reads the Products table back and loads one summary row per product.
    configurator
        .topics()
        .addTopic("product-summary", ITupleTopicConfigurator.class)
        .settings(
            ITupleTopicSettings.of(
                _ -> {
                  final var query =
                      datastore
                          .getQueryManager()
                          .listQuery()
                          .forTable("Products")
                          .withoutCondition()
                          .withTableFields("productId", "productName")
                          .toQuery();
                  final List<Object[]> tuples = new ArrayList<>();
                  // Run the query on the current transaction
                  try (final ICursor cursor =
                      datastore
                          .getTransactionManager()
                          .getQueryRunner()
                          .listQuery(query)
                          .runCurrentThread()) {
                    while (cursor.next()) {
                      final IRecordReader product = cursor.getRecord();
                      tuples.add(new Object[] {product.read(0), product.read(1)});
                    }
                  }
                  return tuples;
                },
                List.of("productId", "productName")))
        .attachChannel("ProductSummaryChannel");
    // Trigger it after the CSV topic so the rows it queries are already in the store.
    configurator.topics().orderTopicAfterTopic("product-summary", "products-csv");
    // Clear ProductSummary before it is re-derived, so its rows do not accumulate across loads.
    configurator
        .triggers()
        .addTrigger(
            "clean-product-summary",
            new IDataOperationTrigger() {
              @Override
              public Set<String> getInputTables() {
                return Set.of("Products");
              }
              @Override
              public Set<String> getOutputTables() {
                return Set.of("ProductSummary");
              }
              @Override
              public void execute(
                  IQueryManager queryManager,
                  IOpenedTransaction transaction,
                  ITriggerContext context) {
                try {
                  transaction.removeWhere("ProductSummary", BaseConditions.TRUE);
                } catch (DatastoreTransactionException e) {
                  throw new RuntimeException(e);
                }
              }
            });
  };
}

The tuple topic re-derives the whole ProductSummary table on every load, so its previous rows must be cleared first, otherwise they accumulate. The example registers a data-operation trigger that empties ProductSummary. Because the trigger reads Products, it is fired by the products-csv topic and runs right after it; and because orderTopicAfterTopic waits for a topic and the triggers it fires, product-summary re-derives only once Products is loaded and ProductSummary has been cleared. For how data-operation triggers work, see data operation triggers.

Load data: general topic, channel, and field calculator concepts.
Load CSV files: file selection, column parsers, and parsing options for the CSV topic.
Data operation triggers: run derived logic inside the load transaction.

​How to load a fixed set of tuples

​How the tuple supplier uses the load scope

​How to build tuples from a datastore query

​Related pages

How to load a fixed set of tuples

How the tuple supplier uses the load scope

How to build tuples from a datastore query

Related pages