# Lookup Catalog Sync Workflow

The Lookup Catalog sync workflow reads from the `cdp_lookup_catalog` database in Data Workbench, detects changed records using hash-based change tracking, and uploads only modified rows to RT 2.0's internal storage via the Bulk Load API.

Workflow files on GitHub
The latest workflow files are available in the [treasure-data/treasure-boxes](https://github.com/treasure-data/treasure-boxes/tree/master/realtime-box/lookup-catalog-sync) repository under `realtime-box/lookup-catalog-sync/`:

* `lookup_catalog_sync.dig` — main workflow
* `scripts.py` — generates type-aware JSON payload SQL per table
* `queries/discover_tables.sql` — discovers eligible tables in `cdp_lookup_catalog`


The workflow automatically discovers all eligible tables in `cdp_lookup_catalog` on each run — there is no separate configuration step per table. An optional `p_table_name` parameter can restrict a single run to one table, which is useful for testing.

## How It Works

The workflow follows the same per-table sync pipeline for every discovered table:

```mermaid
flowchart TD
    A[Discover tables] --> B[For each table]
    B --> C[Extract changed records\nvia hash comparison]
    C --> D{Changes exist?}
    D -->|Yes| E[Upload to RT 2.0\nInternal Storage]
    D -->|No| F[Skip — no update]
    E --> G[Update digest table]
    G --> H[Cleanup temp tables]
```

**Change detection**: Each table has a companion digest table (`_wf_{table_name}_digests`) that stores a hash of each row's JSON payload. On each run, only rows whose hash has changed (or that are new) are uploaded.

**Internal table naming**: All workflow-internal and temporary tables use a `_wf_` prefix to distinguish them from source tables. The `discover_tables.sql` query automatically excludes `_wf_*` tables from sync.

## Configuration Parameters

| **Parameter**  | **Default**  | **Description**  |
|  --- | --- | --- |
| `reactor_importer_endpoint` | (required) | Region-specific Bulk Load API endpoint. See the [region-specific endpoints](#region-specific-endpoints) table below. |
| `reactor_instance` | (required) | Instance identifier provisioned by Treasure AI when RT 2.0 is set up for your account (format: `a{accountId}n{instanceId}`). Contact your Customer Success Manager or Treasure AI Support to obtain this value. |
| `td.database` | `cdp_lookup_catalog` | Source database containing the lookup tables. |
| `p_table_name` | `""` (all tables) | When set, syncs only the specified table. Leave empty to sync all discovered tables. |
| `batch_size` | `1000` | Number of records per upload request to the Bulk Load API. |
| `parallelism` | `10` | Number of concurrent upload threads per table. |


### Region-Specific Endpoints

| **Region**  | **`reactor_importer_endpoint`**  |
|  --- | --- |
| US | `https://bulk-storage-importer-api-production-aws-us-east-1.internal.treasuredata.com` |
| EU (eu01) | `https://bulk-storage-importer-api-production-eu01-eu-central-1.internal.treasuredata.com` |
| Tokyo | `https://bulk-storage-importer-api-production-aws-tokyo-ap-northeast-1.internal.treasuredata.com` |


Internal endpoint access
`.internal.treasuredata.com` endpoints are only reachable from within the Treasure Data platform (for example, from a TD Workflow task) and cannot be accessed from the public internet.

No instance-level data isolation
The `reactor_instance` parameter routes the upload request but does not restrict which instances can read the data. All instances within the account share the same lookup catalog storage — any Lookup Catalog Attribute in any instance can read any table uploaded under the account, regardless of which `reactor_instance` value was used to upload it. If your account operates multiple RT 2.0 instances that must not share catalog data, do not rely on `reactor_instance` for isolation; use separate accounts or contact Treasure AI Support.

## Setting Up the Workflow

1. In Data Workbench, create a new Workflow project (or use an existing one).
2. Copy the `lookup-catalog-sync/` directory from [realtime-box/lookup-catalog-sync/](https://github.com/treasure-data/treasure-boxes/tree/master/realtime-box/lookup-catalog-sync) in the treasure-boxes repository and upload the files (`lookup_catalog_sync.dig`, `scripts.py`, `queries/discover_tables.sql`) into the project.
3. Edit `lookup_catalog_sync.dig` and set `reactor_importer_endpoint`, `reactor_instance`, and `td.database` in the `_export` block. Optionally set `p_table_name` to restrict a run to a single table for testing.
4. Store your TD API key in the project's **Secrets** tab as `td.apikey`. Use a key scoped to the minimum required permission (write access for the Bulk Load API) rather than a full-access key.
5. Set a schedule (daily is recommended) and run the workflow manually once to verify the initial full upload succeeds.


The discovery query automatically finds all base tables in `cdp_lookup_catalog` and excludes `_wf_*` internal tables.

Parallelism guidance
Start with the default settings (`batch_size: 1000`, `parallelism: 10`) for the initial full upload. Increase gradually during incremental runs while monitoring for throttling errors. Larger tables or higher parallelism amplify the risk of throttling.

## Scheduling Recommendations

| **Data type**  | **Recommended schedule**  |
|  --- | --- |
| Flash sale / time-sensitive pricing | Every 1–2 hours (`0 */1 * * *`) |
| Daily promotions / coupon tables | Every 6 hours (`0 */6 * * *`) |
| Product catalog (stable data) | Daily (`0 2 * * *`) |


## Troubleshooting

| **Symptom**  | **Cause and fix**  |
|  --- | --- |
| All records upload on every run | The digest table does not exist or has stale hashes. Drop `_wf_{table_name}_digests` and re-run to rebuild from scratch. |
| 400 error: `"Expected string, received..."` | A row has a `NULL` or empty primary key. Filter out or fix null-key rows in the source table before the next run. |
| 400 error: `"Missing required header: x-lookup-table-name"` | The `x-lookup-table-name` header is missing from the upload step. Verify the workflow template headers are configured correctly. |
| Workflow fails with `varchar = bigint` type error | The source table column type was changed to an incompatible type (for example, `string` → `int`). Drop the digest table, fix the source data, and re-run. |
| Schema mismatch after changing the primary key column | Drop the `_wf_{table_name}_digests` table, then re-run. The workflow will re-upload all rows with the new key. Previously uploaded data under the old key is not automatically deleted. |
| Removed column still appears in responses | RT 2.0 stores data as uploaded — removing a column from the source table does not delete it from internal storage. A full re-upload is required to purge the residual data: drop the digest table (`_wf_{table_name}_digests`) and re-run the workflow. Until then, the deleted column's data remains accessible via the API. |
| Upload timeout | Reduce `parallelism` or `batch_size`, especially during the initial full upload of a large table. |


## Key Column Constraints

The workflow detects the primary key as the **first non-`time` column** in the table schema (by `ordinal_position` in `information_schema.columns`) — if the table's first column is named `time`, the second column is used as the primary key instead.

| **Constraint**  | **Detail**  |
|  --- | --- |
| Column position | The key column must be the first non-`time` column in the table schema. |
| Column name | Must not be named `time` (reserved by Treasure AI platform). |
| Null values | Key column must not contain `NULL` or empty values. Upload fails with 400 if null keys are present. |
| Supported types | `string`. Other types are not supported for the primary key column. |
| Uniqueness | Data must be deduplicated. The workflow does not deduplicate rows. |