Skip to main content
Tutorial

Flatten Order Data into Your Own BigQuery Tables

Overview​

Each order in uc_orders is a single row, with its line items, coupons, and other repeating details stored as arrays inside that row. BI tools, semantic layers, and derived metrics usually want the opposite shape: one flat row per line item. You can flatten with CROSS JOIN UNNEST in every query, and for occasional reports that is the right answer. Once many dashboards and models read the same flattened data, keeping a flat copy in your own Google Cloud project is cheaper and faster.

This tutorial builds that copy. By the end you have:

  • a uc_order_items table in your own project, with one row per order line item
  • a stored procedure that reloads only the orders that changed since its last run
  • a scheduled query that runs the procedure every hour
  • a repeatable way to add columns and more child tables as your needs grow

The example flattens line item cost and quantity, but the same pattern works for any array on an order.

Why not a materialized view​

A BigQuery materialized view cannot read your UltraCart data from your own project. Google requires a materialized view to live in the same project or organization as the tables it reads, and your data warehouse is a project that UltraCart manages. The uc_orders you query is also a view rather than a table, which materialized views do not support. The scheduled refresh in this tutorial gives you the same result in a project you control.

How the refresh works​

Each run asks the streaming table which orders changed since the previous run, then replaces those orders' rows in your table with their current state from uc_orders. Four details make that efficient:

  • partition_date is a weekly bucket. It holds the Sunday that starts the week an order was created. Filtering on it lets BigQuery skip every week that did not change, which is what keeps each run small.
  • Old orders change too. Refunds, shipments, and auto order activity update existing orders, sometimes ones created years ago. On a typical day, changes land in more than a hundred different weeks, so the refresh works out which weeks changed rather than reloading only recent ones.
  • RecordTime marks each change. It records when UltraCart wrote a change into the streaming table. The procedure saves the newest RecordTime it has loaded, called the watermark, in a load_state table, and the next run starts from there.
  • Changed orders are replaced whole. The old rows for an order are deleted and its current rows inserted in one transaction, which handles removed line items and deleted orders without special cases.
note

Data Sets & Streaming tells you not to query the streaming tables, because they hold one row per change rather than one row per order. The refresh follows the spirit of that rule: it reads only partition_date, order_id, and RecordTime from the streaming table to find what changed, and every order value comes from the consistent uc_orders view.

Before you begin​

You need four things in place:

  • A Google Cloud project of your own, with billing enabled. The dataset you create in it must be in the US multi-region, the same location as your UltraCart data warehouse.
  • An account with data warehouse access. This can be your own Google account or a service account. To set up a service account, follow Configuring a User to Run Queries Programmatically.
  • Permissions in your own project to create datasets, tables, and stored procedures, run queries, and create scheduled queries. Scheduled queries also need the BigQuery Data Transfer Service API enabled.
  • Your data warehouse project ID, which follows the pattern ultracart-dw-{merchantid} and appears under Configuration → Developer → Data Warehouse (BigQuery).

Choose your dataset​

The dataset you read from decides which accounts and which columns your tables can hold. Pick one before you start:

  • ultracart_dw holds your primary account's data. Item cost, quantity, and every other column in this tutorial are available here.
  • ultracart_dw_linked holds the data for all your linked accounts in one place. Choose it when your tables should include orders from every linked account.
  • ultracart_dw_medium or higher, or ultracart_dw_linked_medium or higher, is only needed when you add columns containing customer personal information, such as billing names or email addresses.

Each dataset reads from a streaming dataset, and the procedure checks that streaming dataset for changes. Datasets whose names start with ultracart_dw_linked use ultracart_dw_linked_streaming. All the others use ultracart_dw_streaming. The access level each dataset requires is listed under Data Sets & Streaming.

The SQL on this page uses placeholders. Replace each one everywhere it appears:

  • your-project becomes your own Google Cloud project ID.
  • ultracart-dw-yourmerchantid becomes your data warehouse project ID, in lowercase.
  • ultracart_dw.uc_orders becomes your chosen dataset, for example ultracart_dw_linked.uc_orders.
  • ultracart_dw_streaming.uc_order_streaming becomes ultracart_dw_linked_streaming.uc_order_streaming if you chose a linked dataset.

Step 1: Create the dataset and tables​

Run this in the BigQuery console, in your own project. It creates a dataset named ultracart_flat, a load_state table that tracks each flat table's watermark, and the uc_order_items table itself.

CREATE SCHEMA IF NOT EXISTS `your-project.ultracart_flat`
OPTIONS (location = 'US');

-- One row per flattened table: the newest UltraCart change (RecordTime) that has been loaded.
CREATE TABLE IF NOT EXISTS `your-project.ultracart_flat.load_state` (
  table_name  STRING NOT NULL,
  watermark   DATETIME NOT NULL,
  updated_at  TIMESTAMP NOT NULL
);

-- One row per order line item.
CREATE TABLE IF NOT EXISTS `your-project.ultracart_flat.uc_order_items` (
  merchant_id                     STRING,
  order_id                        STRING NOT NULL,
  partition_date                  DATE,            -- Sunday of the week the order was created
  item_position                   INT64 NOT NULL,  -- 0-based position in the order's items array
  item_index                      INT64,           -- UltraCart item index (empty on some older orders)
  merchant_item_id                STRING,
  description                     STRING,
  quantity                        NUMERIC,
  kit                             BOOL,
  kit_component                   BOOL,
  currency_code                   STRING,          -- from the order
  cost_value                      NUMERIC,
  cost_localized                  NUMERIC,
  cost_localized_formatted        STRING,
  cost_exchange_rate              NUMERIC,
  discount_value                  NUMERIC,
  total_cost_with_discount_value  NUMERIC,
  cogs                            NUMERIC,
  quantity_refunded               NUMERIC,
  total_refunded_value            NUMERIC
)
PARTITION BY partition_date
CLUSTER BY order_id;

The table is partitioned by the same partition_date as uc_orders and clustered by order_id, so the refresh's delete step only touches the weeks and orders it is replacing.

Step 2: Create the refresh procedure​

The procedure does all the work, and it runs in one of two modes. Called with TRUE, it rebuilds the whole table. Called with FALSE, it reloads only the orders that changed since the last run.

CREATE OR REPLACE PROCEDURE `your-project.ultracart_flat.refresh_uc_order_items`(full_reload BOOL)
BEGIN
  DECLARE last_watermark DATETIME;
  DECLARE new_watermark DATETIME;
  DECLARE changed_partitions ARRAY<DATE>;
  DECLARE changed_orders ARRAY<STRING>;

  -- 1. Work out what to reload. This reads only partition_date, order_id and RecordTime
  --    from the streaming table, so it is cheap and touches no customer data.
  IF full_reload THEN
    SET (changed_partitions, new_watermark) = (
      SELECT AS STRUCT ARRAY_AGG(DISTINCT partition_date IGNORE NULLS), MAX(RecordTime)
      FROM `ultracart-dw-yourmerchantid.ultracart_dw_streaming.uc_order_streaming`
    );
  ELSE
    SET last_watermark = (
      SELECT watermark
      FROM `your-project.ultracart_flat.load_state`
      WHERE table_name = 'uc_order_items'
    );
    IF last_watermark IS NULL THEN
      RAISE USING MESSAGE = 'uc_order_items has never been loaded. Run CALL refresh_uc_order_items(TRUE) first.';
    END IF;

    -- Look back 15 minutes before the last watermark in case changes arrived out of order.
    -- Reloading an order twice is harmless.
    SET (changed_partitions, changed_orders, new_watermark) = (
      SELECT AS STRUCT
        ARRAY_AGG(DISTINCT partition_date IGNORE NULLS),
        ARRAY_AGG(DISTINCT order_id IGNORE NULLS),
        MAX(RecordTime)
      FROM `ultracart-dw-yourmerchantid.ultracart_dw_streaming.uc_order_streaming`
      WHERE RecordTime > DATETIME_SUB(last_watermark, INTERVAL 15 MINUTE)
    );

    IF new_watermark IS NULL THEN
      RETURN;  -- nothing changed
    END IF;
  END IF;

  -- 2. Replace the rows for those orders. All or nothing.
  BEGIN
    BEGIN TRANSACTION;

    DELETE FROM `your-project.ultracart_flat.uc_order_items`
    WHERE full_reload
       OR (partition_date IN UNNEST(changed_partitions) AND order_id IN UNNEST(changed_orders));

    -- The column list is explicit so that adding a column to the table never breaks this INSERT.
    INSERT INTO `your-project.ultracart_flat.uc_order_items` (
      merchant_id,
      order_id,
      partition_date,
      item_position,
      item_index,
      merchant_item_id,
      description,
      quantity,
      kit,
      kit_component,
      currency_code,
      cost_value,
      cost_localized,
      cost_localized_formatted,
      cost_exchange_rate,
      discount_value,
      total_cost_with_discount_value,
      cogs,
      quantity_refunded,
      total_refunded_value
    )
    SELECT
      o.merchant_id,
      o.order_id,
      o.partition_date,
      item_position,
      i.item_index,
      i.merchant_item_id,
      i.description,
      i.quantity,
      i.kit,
      i.kit_component,
      o.currency_code,
      i.cost.value,
      i.cost.localized,
      i.cost.localized_formatted,
      i.cost.exchange_rate,
      i.discount.value,
      i.total_cost_with_discount.value,
      i.cogs,
      i.quantity_refunded,
      i.total_refunded.value
    FROM `ultracart-dw-yourmerchantid.ultracart_dw.uc_orders` AS o
    CROSS JOIN UNNEST(o.items) AS i WITH OFFSET AS item_position
    -- The partition_date filter keeps this cheap: BigQuery only reads the weeks that changed.
    WHERE o.partition_date IN UNNEST(changed_partitions)
      AND (full_reload OR o.order_id IN UNNEST(changed_orders));

    MERGE `your-project.ultracart_flat.load_state` AS s
    USING (SELECT 'uc_order_items' AS table_name, new_watermark AS watermark) AS n
    ON s.table_name = n.table_name
    WHEN MATCHED THEN
      UPDATE SET watermark = GREATEST(s.watermark, n.watermark), updated_at = CURRENT_TIMESTAMP()
    WHEN NOT MATCHED THEN
      INSERT (table_name, watermark, updated_at) VALUES (n.table_name, n.watermark, CURRENT_TIMESTAMP());

    COMMIT TRANSACTION;
  EXCEPTION WHEN ERROR THEN
    ROLLBACK TRANSACTION;
    RAISE USING MESSAGE = FORMAT('refresh_uc_order_items failed and was rolled back: %s', @@error.message);
  END;
END;

A few choices in this procedure are worth understanding before you change it:

  • The 15-minute look-back re-checks changes just before the watermark. A change that reached the streaming table slightly out of order is still picked up, and an order reloaded twice ends up the same.
  • The transaction makes each run all or nothing. If any statement fails, the deletes, inserts, and watermark update are all rolled back, and the next run tries the same changes again.
  • The nested cost record becomes plain columns. It holds one value per item, so its fields (cost.value, cost.localized) sit on the item row as cost_value and cost_localized rather than in a table of their own.

Step 3: Load the table for the first time​

Run the procedure once in full reload mode:

CALL `your-project.ultracart_flat.refresh_uc_order_items`(TRUE);

A full reload reads every order once, so its cost grows with your order history. On an account with about four million orders, the columns in this tutorial read about 1.8 GB. To estimate yours first, paste the procedure's SELECT ... FROM ... CROSS JOIN UNNEST(...) into the BigQuery console without its WHERE clause, and read the byte estimate the console shows before you run anything.

Step 4: Schedule the refresh​

A scheduled query that calls the procedure every hour keeps your table current:

  1. In the BigQuery console for your own project, open a new query and enter:

    CALL `your-project.ultracart_flat.refresh_uc_order_items`(FALSE);

  2. Select Schedule and give the scheduled query a name, such as Refresh uc_order_items.

  3. Set it to repeat every hour. Leave the destination table empty, because the procedure writes to its own tables.

  4. If you set up a service account for your data warehouse access, choose it as the account the query runs as, then save.

Google's scheduling queries guide covers each option on that screen.

Hourly runs stay small because only the weeks that changed are read. On the four-million-order account above, an hourly run typically reads a few hundred megabytes or less for the reload, plus about 15 MB to find the changes. Running more often costs little more, since each run has fewer changes to process.

Step 5: Check the result​

Compare row counts between your table and the uc_orders view for recent weeks:

WITH source AS (
  SELECT o.partition_date, COUNT(*) AS source_items
  FROM `ultracart-dw-yourmerchantid.ultracart_dw.uc_orders` AS o
  CROSS JOIN UNNEST(o.items) AS i
  WHERE o.partition_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 8 WEEK)
  GROUP BY o.partition_date
),
flat AS (
  SELECT partition_date, COUNT(*) AS flat_items
  FROM `your-project.ultracart_flat.uc_order_items`
  WHERE partition_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 8 WEEK)
  GROUP BY partition_date
)
SELECT
  partition_date,
  source.source_items,
  flat.flat_items,
  COALESCE(source.source_items, 0) - COALESCE(flat.flat_items, 0) AS difference
FROM source
FULL OUTER JOIN flat USING (partition_date)
ORDER BY partition_date DESC;

Every difference should be 0, apart from small gaps in the current week from orders placed since the last scheduled run.

Add a column​

Adding a column takes three steps, and your scheduled refresh keeps running throughout. This example adds the shipped date and item weight.

  1. Add the columns to the table. Until the procedure is updated, scheduled runs leave them empty.

    ALTER TABLE `your-project.ultracart_flat.uc_order_items`
      ADD COLUMN IF NOT EXISTS shipped_dts DATETIME,
      ADD COLUMN IF NOT EXISTS weight_value NUMERIC,
      ADD COLUMN IF NOT EXISTS weight_uom STRING;

  2. Edit the procedure from Step 2. Add shipped_dts, weight_value, weight_uom to the end of the INSERT column list, and i.shipped_dts, i.weight.value, i.weight.uom to the end of the SELECT, in the same order. Run the whole CREATE OR REPLACE PROCEDURE statement again.

  3. Fill the new columns for the orders already in the table:

    CALL `your-project.ultracart_flat.refresh_uc_order_items`(TRUE);

To find the names of the fields you can add, open uc_orders in the BigQuery console and expand items on the Schema tab.

Add another child table​

Arrays are the parts of an order that need a table of their own. A nested record such as cost holds one value per item and becomes columns, as in Step 2. An array such as coupons, utms, or current_stage_histories holds any number of entries per order, so each one becomes its own table with its own procedure.

This example adds uc_order_coupons. It follows Step 1 and Step 2 exactly, with a different table name, a different array in the CROSS JOIN UNNEST, and its own row in load_state:

CREATE TABLE IF NOT EXISTS `your-project.ultracart_flat.uc_order_coupons` (
  merchant_id            STRING,
  order_id               STRING NOT NULL,
  partition_date         DATE,
  coupon_position        INT64 NOT NULL,
  coupon_code            STRING,
  base_coupon_code       STRING,
  accounting_code        STRING,
  automatically_applied  BOOL
)
PARTITION BY partition_date
CLUSTER BY order_id;

CREATE OR REPLACE PROCEDURE `your-project.ultracart_flat.refresh_uc_order_coupons`(full_reload BOOL)
BEGIN
  DECLARE last_watermark DATETIME;
  DECLARE new_watermark DATETIME;
  DECLARE changed_partitions ARRAY<DATE>;
  DECLARE changed_orders ARRAY<STRING>;

  IF full_reload THEN
    SET (changed_partitions, new_watermark) = (
      SELECT AS STRUCT ARRAY_AGG(DISTINCT partition_date IGNORE NULLS), MAX(RecordTime)
      FROM `ultracart-dw-yourmerchantid.ultracart_dw_streaming.uc_order_streaming`
    );
  ELSE
    SET last_watermark = (
      SELECT watermark
      FROM `your-project.ultracart_flat.load_state`
      WHERE table_name = 'uc_order_coupons'
    );
    IF last_watermark IS NULL THEN
      RAISE USING MESSAGE = 'uc_order_coupons has never been loaded. Run CALL refresh_uc_order_coupons(TRUE) first.';
    END IF;

    SET (changed_partitions, changed_orders, new_watermark) = (
      SELECT AS STRUCT
        ARRAY_AGG(DISTINCT partition_date IGNORE NULLS),
        ARRAY_AGG(DISTINCT order_id IGNORE NULLS),
        MAX(RecordTime)
      FROM `ultracart-dw-yourmerchantid.ultracart_dw_streaming.uc_order_streaming`
      WHERE RecordTime > DATETIME_SUB(last_watermark, INTERVAL 15 MINUTE)
    );

    IF new_watermark IS NULL THEN
      RETURN;
    END IF;
  END IF;

  BEGIN
    BEGIN TRANSACTION;

    DELETE FROM `your-project.ultracart_flat.uc_order_coupons`
    WHERE full_reload
       OR (partition_date IN UNNEST(changed_partitions) AND order_id IN UNNEST(changed_orders));

    INSERT INTO `your-project.ultracart_flat.uc_order_coupons` (
      merchant_id,
      order_id,
      partition_date,
      coupon_position,
      coupon_code,
      base_coupon_code,
      accounting_code,
      automatically_applied
    )
    SELECT
      o.merchant_id,
      o.order_id,
      o.partition_date,
      coupon_position,
      c.coupon_code,
      c.base_coupon_code,
      c.accounting_code,
      c.automatically_applied
    FROM `ultracart-dw-yourmerchantid.ultracart_dw.uc_orders` AS o
    CROSS JOIN UNNEST(o.coupons) AS c WITH OFFSET AS coupon_position
    WHERE o.partition_date IN UNNEST(changed_partitions)
      AND (full_reload OR o.order_id IN UNNEST(changed_orders));

    MERGE `your-project.ultracart_flat.load_state` AS s
    USING (SELECT 'uc_order_coupons' AS table_name, new_watermark AS watermark) AS n
    ON s.table_name = n.table_name
    WHEN MATCHED THEN
      UPDATE SET watermark = GREATEST(s.watermark, n.watermark), updated_at = CURRENT_TIMESTAMP()
    WHEN NOT MATCHED THEN
      INSERT (table_name, watermark, updated_at) VALUES (n.table_name, n.watermark, CURRENT_TIMESTAMP());

    COMMIT TRANSACTION;
  EXCEPTION WHEN ERROR THEN
    ROLLBACK TRANSACTION;
    RAISE USING MESSAGE = FORMAT('refresh_uc_order_coupons failed and was rolled back: %s', @@error.message);
  END;
END;

Then load it with CALL ...refresh_uc_order_coupons(TRUE) and schedule CALL ...refresh_uc_order_coupons(FALSE), the same way as Steps 3 and 4.

Arrays inside a line item, such as items.options or items.properties, work the same way with a second UNNEST. The key for each row becomes the order, the item position, and the option position:

FROM `ultracart-dw-yourmerchantid.ultracart_dw.uc_orders` AS o
CROSS JOIN UNNEST(o.items) AS i WITH OFFSET AS item_position
CROSS JOIN UNNEST(i.options) AS opt WITH OFFSET AS option_position

Things to know about line item rows​

A few properties of the item data affect how you build metrics on top of uc_order_items:

  • Use order_id and item_position as the row key. item_index is UltraCart's own item number and is unique within an order when present, but it is empty on some older orders.
  • Take currency from the order. items.cost.currency_code can be empty on every row of an account, which is why the table stores the order-level currency_code instead.
  • Kit components are rows of their own. A kit and each of its components appear as separate items (kit and kit_component). Check how your kits split cost between the parent and its components before summing cost or quantity, so nothing is counted twice.

Troubleshooting​

The procedure says the table has never been loaded​

The scheduled query ran before the first full reload, so load_state has no watermark for the table. Run CALL ...refresh_uc_order_items(TRUE) once, and the next scheduled run succeeds.

A run fails because of a concurrent transaction​

Two refreshes of the same table ran at the same time, most often a manual full reload that overlapped a scheduled run. BigQuery cancels one of the conflicting transactions and rolls it back, so no data is lost. Run it again once the other has finished.

Access Denied on the data warehouse project​

The account running the query lacks access to the dataset or streaming dataset the procedure reads. The account owner grants data warehouse access on the User Configuration Screen. A scheduled query runs as the account chosen when it was created, which may differ from the account you tested with.

A table is reported as not found in location US​

BigQuery raises this error when the dataset in your project was created outside the US multi-region. BigQuery cannot move a dataset, so create a new one with OPTIONS (location = 'US'), rerun Steps 1 and 2 against it, and run a full reload.

Deleted orders still appear​

UltraCart periodically clears old rows from the streaming tables, including the markers that record a deleted order. If your scheduled query was paused or failing long enough for those markers to be cleared, the refresh never sees the deletion. Run a full reload to bring the table back in line, and keep the schedule at every three hours or more often.

Next steps​

Was this page helpful?