Lookup Tables

Lookup tables provide a centralized database solution for actions requiring tabular data. A simple API in the DeltaFi Action Kit provides queries for the data.

Configuration (Kubernetes)

Enabling lookup tables in a RKE2 (Kubernetes) orchestrated environment requires a persistent volume (pv) to be configured and available for use prior to the service being enabled.

Adding the persistent volume below to the inventory file used in Ansible Setup and rerunning booststrap-deltafi

yaml

---
k8s_cluster:
  children:
    masters:
      hosts:
        <host>:
          persistent_volumes:
            - name: deltafi-postgres-lookup
              path: /data/deltafi/postgres-lookup
              storage_class: deltafi-postgres-lookup
              size: 10G

bash

cd ~/ansible
ansible-playbook bootstrap-deltafi.yml -i inventory/site

Once the persistent volume is available update your deltafi/site/values.yaml to enable the service and run deltafi up

yaml

deltafi:
  lookup:
    enabled: true

Lookup Table Suppliers

Lookup table suppliers are defined in Plugins to supply data to lookup tables. Suppliers will provide data on startup and can be configured to refresh it periodically.

Java

A lookup table supplier will extend the LookupTableSupplier class, providing the LookupTableClient (auto-configured by the DeltaFi Action Kit) and the lookup table definition. It will also define the getRows method to provide rows for the lookup table.

java

package org.deltafi.helloworld;

import lombok.extern.slf4j.Slf4j;
import org.deltafi.actionkit.lookup.LookupTableClient;
import org.deltafi.actionkit.lookup.LookupTableSupplier;
import org.deltafi.common.lookup.LookupTable;
import org.jetbrains.annotations.Nullable;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
@Slf4j
public class MyLookupTableSupplier extends LookupTableSupplier {
    public MyLookupTableSupplier(LookupTableClient lookupTableClient) {
        super(lookupTableClient, LookupTable.builder()
                .name("my_lookup_table")
                .columns(List.of("column_a", "column_b", "color"))
                .keyColumns(List.of("column_a"))
                .refreshDuration("PT1M")
                .build());
    }

    @Override
    public List<Map<String, String>> getRows(@Nullable Map<String, Set<String>> matchingColumnValues,
            @Nullable List<String> resultColumns) {
        log.info("Getting rows for {}", getLookupTable().getName());

        return List.of(
                Map.of("column_a", "A1", "column_b", "B1", "color", randomColor()),
                Map.of("column_a", "A2", "column_b", "B2", "color", randomColor()),
                Map.of("column_a", "A3", "column_b", "B3", "color", randomColor()));
    }

    private static final List<String> COLORS = List.of("red", "orange", "yellow", "green", "blue", "indigo", "violet");

    private String randomColor() {
        return COLORS.get((int) (Math.random() * COLORS.size()));
    }
}

Go

A lookup table supplier in Go implements the LookupTableSupplier interface, which defines the table schema and provides rows on demand.

package suppliers

import (
    "context"
    "log"
    "math/rand/v2"

    actionkit "gitlab.com/deltafi/deltafi/deltafi-go-action-kit/v2"
)

var colors = []string{"red", "orange", "yellow", "green", "blue", "indigo", "violet"}

func init() {
    actionkit.RegisterSupplier(&MyLookupTableSupplier{})
}

type MyLookupTableSupplier struct{}

func (s *MyLookupTableSupplier) Table() actionkit.LookupTableDefinition {
    return actionkit.LookupTableDefinition{
        Name:            "my_lookup_table",
        Columns:         []string{"column_a", "column_b", "color"},
        KeyColumns:      []string{"column_a"},
        RefreshDuration: "PT1M",
    }
}

func (s *MyLookupTableSupplier) GetRows(ctx context.Context, variables map[string]string,
    matchingColumns map[string][]string, resultColumns []string) ([]actionkit.Row, error) {
    log.Println("Getting rows for my_lookup_table")
    return []actionkit.Row{
        actionkit.NewRow(map[string]string{"column_a": "A1", "column_b": "B1", "color": colors[rand.IntN(len(colors))]}),
        actionkit.NewRow(map[string]string{"column_a": "A2", "column_b": "B2", "color": colors[rand.IntN(len(colors))]}),
        actionkit.NewRow(map[string]string{"column_a": "A3", "column_b": "B3", "color": colors[rand.IntN(len(colors))]}),
    }, nil
}

C++

A lookup table supplier in C++ extends LookupTableSupplierBase, which defines the table schema and provides rows on demand.

cpp

#pragma once
#include <deltafi/plugin>

class MyLookupTableSupplier : public deltafi::LookupTableSupplierBase {
public:
    deltafi::LookupTableDefinition table() const override {
        return {
            .name = "my_lookup_table",
            .columns = {"column_a", "column_b", "color"},
            .key_columns = {"column_a"},
            .refresh_duration = "PT1M",
        };
    }

    std::vector<std::map<std::string, std::string>> get_rows(
        const std::map<std::string, std::string>& /*variables*/,
        const std::map<std::string, std::vector<std::string>>& /*matching_column_values*/,
        const std::vector<std::string>& /*result_columns*/) override {
        return {
            {{"column_a", "A1"}, {"column_b", "B1"}, {"color", "red"}},
            {{"column_a", "A2"}, {"column_b", "B2"}, {"color", "green"}},
            {{"column_a", "A3"}, {"column_b", "B3"}, {"color", "blue"}},
        };
    }
};

DELTAFI_SUPPLIER(MyLookupTableSupplier)

Lookup Table API

Java

The LookupTableClient (auto-configured by the DeltaFi Action Kit) provides a lookup method for Actions to query a lookup table. First, pass the LookupTableClient to the Action constructor and assign it to an instance variable. Then, use it as follows:

...
    try {
        LookupResults rgbRows = lookupTableClient.lookup("my_lookup_table",
                LookupOptions.builder()
                    .matchingColumnValues(Map.of("color", Set.of("red", "green", "blue")))
                    .resultColumns(Set.of("column_a", "color"))
                    .sortColumn("column_a")
                    .sortDirection(SortDirection.DESC)
                    .offset(offset)
                    .limit(limit)
                    .build());
        result.addMetadata("Total RGB Rows", rgbRows.totalCount());
        result.addMetadata("RGB Rows", String.join(", ", rgbRows.results().stream().map(m -> m.get("column_a")).toList()));
    } catch (Exception e) {
        return new ErrorResult(context, "Unable to lookup RGB rows", e);
    }

To return all (unsorted) rows with all result columns, pass LookupOptions.defaultLookupOptions().

Go

The package-level Lookup function queries lookup tables using a builder pattern:

func (a *MyAction) Transform(df *actionkit.Deltafile, params MyParams) *actionkit.Deltafile {
    lookupResults, err := actionkit.Lookup("my_lookup_table",
        actionkit.LookupQuery().
            WithMatch("color", "red", "green", "blue").
            WithResultColumns("column_a", "color").
            WithSort("column_a", actionkit.SortDESC).
            WithOffset(0).
            WithLimit(100))
    if err != nil {
        return df.Errorf("Unable to lookup rows").SetErrorContext(err.Error())
    }

    df.PassthroughContent()
    df.AddMetadata("totalRows", fmt.Sprintf("%d", lookupResults.TotalCount))
    // lookupResults.Rows is []actionkit.Row; access values with Row.Get(column)
    for _, row := range lookupResults.Rows {
        if value, ok := row.Get("column_a"); ok {
            df.LogInfo("column_a: %s", value)
        }
    }
    return df
}

To return all (unsorted) rows with all result columns, pass nil for the query builder.

C++

The LookupClient is available on the ActionContext, which a transform (or join) action reaches with df.context().lookup_client, and provides a lookup method for querying lookup tables:

cpp

deltafi::DeltaFile& transform(deltafi::DeltaFile& df) {
    deltafi::LookupOptions opts;
    opts.matching_column_values = {{"color", {"red", "green", "blue"}}};
    opts.result_columns = {"column_a", "color"};
    opts.sort_column = "column_a";
    opts.sort_direction = deltafi::SortDirection::DESC;
    opts.offset = 0;
    opts.limit = 100;

    try {
        auto results = df.context().lookup_client->lookup("my_lookup_table", &opts);
        df.passthrough_content();
        df.add_metadata("totalRows", std::to_string(results.total_count));
        // results.rows is std::vector<std::map<std::string, std::string>>
        for (const auto& row : results.rows) {
            auto it = row.find("column_a");
            if (it != row.end()) {
                df.log_info("column_a: " + it->second);
            }
        }
    } catch (const std::exception& ex) {
        return df.error("Unable to lookup rows", ex.what());
    }
    return df;
}

To return all (unsorted) rows with all result columns, pass nullptr as the options.

Query results are cached; concurrent lookups for the same key coalesce into a single backend call (single-flight), and expired entries are evicted on a background timer. Each result is cached for results.ttl_seconds, the TTL core returns with it (see Result Caching); a result without a positive ttl_seconds is not cached. The optional TABLE_LOOKUP_CACHE_TTL environment variable (seconds) caps that TTL; it is unset by default and never enables caching a result core sent without a TTL.

The same client also provides admin operations (matching the Java kit):

cpp

auto& lookup = *df.context().lookup_client;
auto tables = lookup.get_lookup_tables();

deltafi::LookupTableDefinition def;
def.name = "colors";
def.columns = {"name", "hex"};
def.key_columns = {"name"};
lookup.create_lookup_table(def);

std::vector<std::map<std::string, std::string>> rows = {{{"name", "red"}, {"hex", "#f00"}}};
lookup.upsert_rows("colors", rows);      // batched at 10,000 rows/request
lookup.remove_rows("colors", {{{"name", "red"}}});
lookup.set_backing_service_active("colors", true);

Each admin call returns a LookupAdminResult (success, info, errors).

Result Caching

Core decides how long a lookup result may be cached and returns that as ttlSeconds on the result:

Table	`ttlSeconds`
Has a `refreshDuration` and a supplier	Seconds until the next scheduled supplier refresh
Has a `refreshDuration`, no supplier	The `refreshDuration`
No `refreshDuration`	The `defaultLookupCacheTtl` system property (default `PT30S`)
Pull-through	Not set: every lookup goes to the supplier
Supplier refresh already overdue	Not set: the next refresh is expected at any moment

The Java and C++ action kits cache each result for its ttlSeconds and do not cache a result without one. Setting defaultLookupCacheTtl to 0 stops these kits from caching lookups against tables without a refreshDuration, so every such lookup queries core. Any other value must be at least one second; include a unit (30s or PT30S), since a bare number is read as milliseconds.

Changes to a cached table become visible to a Java or C++ action once its cached result expires. This includes rows the action itself writes with upsertRows or removeRows (upsert_rows or remove_rows in C++): an action that must read its own writes immediately can call clearCache() (clear_cache() in C++) on the client.

TIP

The Go action kit caches every result for a fixed TABLE_LOOKUP_CACHE_TTL (default 30 seconds) and the Python action kit does not cache, so defaultLookupCacheTtl does not affect those kits.

Lookup Tables ​

Configuration (Kubernetes) ​

Lookup Table Suppliers ​

Java ​

Go ​

C++ ​

Lookup Table API ​

Java ​

Go ​

C++ ​

Result Caching ​

Lookup Tables

Configuration (Kubernetes)

Lookup Table Suppliers

Java

Go

C++

Lookup Table API

Java

Go

C++

Result Caching