Serving data from a lakehouse: grain and the join that counts revenue twice

Module 7 built the shop's pipeline, from order files to a silver table of 1 row per order. This module serves it: it gives each kind of reader the data in the way they read it. This part needs module 4's delta_session.py, the helper that starts a local Spark session with Delta Lake.

4 readers, 4 ways in

A lakehouse has more readers than the people who built it, and each one reads differently:

All 4 start from the same few gold tables, the serving tables. If a serving table is wrong, every way in is wrong in the same way. So this part builds one and makes sure it is right.

The grain comes first

A serving table's grain says what 1 row is. The shop's is "1 row per sold order", where sold means not cancelled and not refunded. Readers rely on the grain without knowing it: a dashboard that sums amount is correct only if each order is there once.

Save this as setup_serving.py. It writes the 2 silver tables this module serves from: orders as module 7 left it, and customer_versions, a type 2 table of where each customer lived, as in module 2. Customer C01 moved from Portugal to Spain on 2026-10-02.

"""Write the 2 silver tables this module serves from: orders, and the customers' versions."""
from delta_session import delta_spark

spark = delta_spark("setup")

# Silver orders as module 7's pipeline left them: 1 row per order, in its newest good state.
spark.sql("DROP TABLE IF EXISTS orders")
spark.sql("""
CREATE TABLE orders (order_id BIGINT, customer_id STRING, status STRING, amount DECIMAL(10,2),
                     updated_at TIMESTAMP, placed_at TIMESTAMP) USING delta
""")
spark.sql("""
INSERT INTO orders VALUES
  (1001, 'C01', 'shipped',   25.50, TIMESTAMP'2026-10-02 08:00:00', TIMESTAMP'2026-10-01 09:15:00'),
  (1002, 'C02', 'cancelled', 12.00, TIMESTAMP'2026-10-03 09:30:00', TIMESTAMP'2026-10-01 10:02:00'),
  (1004, 'C01', 'shipped',   40.00, TIMESTAMP'2026-10-04 09:00:00', TIMESTAMP'2026-10-02 12:45:00'),
  (1005, 'C02', 'placed',    18.25, TIMESTAMP'2026-10-03 16:20:00', TIMESTAMP'2026-10-03 16:20:00'),
  (1007, 'C01', 'refunded',   9.99, TIMESTAMP'2026-10-04 11:00:00', TIMESTAMP'2026-10-04 11:00:00')
""")

# 1 row per version of a customer, as module 2's type 2 dimension keeps them. C01 moved on 2026-10-02.
spark.sql("DROP TABLE IF EXISTS customer_versions")
spark.sql("""
CREATE TABLE customer_versions (customer_id STRING, country STRING,
                                valid_from TIMESTAMP, valid_to TIMESTAMP) USING delta
""")
spark.sql("""
INSERT INTO customer_versions VALUES
  ('C01', 'PT', TIMESTAMP'1900-01-01 00:00:00', TIMESTAMP'2026-10-02 00:00:00'),
  ('C01', 'ES', TIMESTAMP'2026-10-02 00:00:00', TIMESTAMP'9999-12-31 00:00:00'),
  ('C02', 'IE', TIMESTAMP'1900-01-01 00:00:00', TIMESTAMP'9999-12-31 00:00:00'),
  ('C03', 'DE', TIMESTAMP'1900-01-01 00:00:00', TIMESTAMP'9999-12-31 00:00:00')
""")
print("orders", spark.table("orders").count(), "rows; customer_versions", spark.table("customer_versions").count(), "rows")

Then save run_sql.py, which runs each statement of a SQL file and shows what a SELECT returns. On Databricks, run each statement in a SQL cell of a notebook instead.

"""Run each statement of the SQL files named, and show what a SELECT returns."""
import sys
from pathlib import Path
from delta_session import delta_spark

spark = delta_spark("sql")
for path in sys.argv[1:]:
    for statement in Path(path).read_text().split(";"):
        lines = [line for line in statement.splitlines() if line.strip() and not line.strip().startswith("--")]
        if not lines:
            continue
        result = spark.sql("\n".join(lines))
        if lines[0].lstrip().upper().startswith(("SELECT", "WITH")):
            result.show(truncate=False)

Run python setup_serving.py. It prints orders 5 rows; customer_versions 4 rows.

The join that counts twice

An analyst wants revenue by country. Save this as fanout.sql:

-- Revenue by the customer's country: the first try.
SELECT c.country, count(*) AS orders, sum(o.amount) AS revenue
FROM orders o
JOIN customer_versions c ON o.customer_id = c.customer_id
WHERE o.status NOT IN ('cancelled', 'refunded')
GROUP BY c.country
ORDER BY c.country;

SELECT sum(amount) AS revenue
FROM orders
WHERE status NOT IN ('cancelled', 'refunded');

python run_sql.py fanout.sql prints:

+-------+------+-------+
|country|orders|revenue|
+-------+------+-------+
|ES     |2     |65.50  |
|IE     |1     |18.25  |
|PT     |2     |65.50  |
+-------+------+-------+

+-------+
|revenue|
+-------+
|83.75  |
+-------+

The countries add up to 149.25, but the shop sold 83.75. C01 has 2 versions, so each of C01's 2 orders matched both and was counted for Portugal and for Spain. Nothing failed and no customer is missing, which is why this kind of mistake reaches dashboards. Switch the join below to see the rows it makes:

Try itThe join that counts an order twice

Join on

Customer onlyVersion at the order

ON o.customer_id = c.customer_id

Rows after the join

order_id	placed	country	amount
1001	2026-10-01	PT	25.50
1001	2026-10-01	ES	25.50
1004	2026-10-02	PT	40.00
1004	2026-10-02	ES	40.00
1005	2026-10-03	IE	18.25

Revenue by country

ES65.50

IE18.25

PT65.50

The countries add up to 149.25; the orders sold 83.75.78% too high

Join each order to the version that was true when it was placed, and each order matches exactly 1 row. Save this as asof.sql:

-- Revenue by the country the customer lived in when they placed the order.
SELECT c.country, count(*) AS orders, sum(o.amount) AS revenue
FROM orders o
JOIN customer_versions c
  ON o.customer_id = c.customer_id
 AND o.placed_at >= c.valid_from
 AND o.placed_at < c.valid_to
WHERE o.status NOT IN ('cancelled', 'refunded')
GROUP BY c.country
ORDER BY c.country;

+-------+------+-------+
|country|orders|revenue|
+-------+------+-------+
|ES     |1     |40.00  |
|IE     |1     |18.25  |
|PT     |1     |25.50  |
+-------+------+-------+

Order 1001, placed on 2026-10-01, counts for Portugal, and 1004, placed after the move, for Spain. The countries add up to 83.75.

ScenarioThe revenue dashboard that is 17% too highApril revenue on the sales dashboard is 17.2% above finance's figure, and the board meeting is at 14:00. Find the cause in the query, not in the warehouse.

Build the serving table, then let readers see it

Rather than leave every analyst to write that join, put it in a serving table once, and check it before it replaces the version readers see. Save this as sales.sql:

-- The rows of the serving table: 1 row per sold order, with the country the customer lived in at the time.
CREATE OR REPLACE TEMPORARY VIEW sales_next AS
SELECT o.order_id,
       to_date(o.placed_at) AS day,
       o.customer_id,
       c.country,
       o.amount
FROM orders o
LEFT JOIN customer_versions c
  ON o.customer_id = c.customer_id
 AND o.placed_at >= c.valid_from
 AND o.placed_at < c.valid_to
WHERE o.status NOT IN ('cancelled', 'refunded');

-- Each check returns NULL when it passes, and stops the script with its message when it fails.
SELECT assert_true(count(*) = count(DISTINCT order_id), 'sales has an order more than once') AS one_row_per_order,
       assert_true(count(*) = count(country), 'a sold order has no country') AS every_order_has_a_country,
       assert_true(sum(amount) = (SELECT sum(amount) FROM orders WHERE status NOT IN ('cancelled', 'refunded')),
                   'sales does not add up to the sold orders') AS revenue_adds_up
FROM sales_next;

-- Only rows that passed every check replace what the readers see.
CREATE OR REPLACE TABLE sales USING delta
COMMENT 'Sold orders, 1 row per order. Cancelled and refunded orders are left out.'
AS SELECT * FROM sales_next;

ALTER TABLE sales ALTER COLUMN day COMMENT 'The day the order was placed, in UTC';
ALTER TABLE sales ALTER COLUMN country COMMENT 'Where the customer lived when they placed the order';
ALTER TABLE sales ALTER COLUMN amount COMMENT 'What the customer paid, in euros';

SELECT * FROM sales ORDER BY order_id;

python run_sql.py sales.sql prints 3 passing checks, then the table:

+-----------------+-------------------------+---------------+
|one_row_per_order|every_order_has_a_country|revenue_adds_up|
+-----------------+-------------------------+---------------+
|NULL             |NULL                     |NULL           |
+-----------------+-------------------------+---------------+

+--------+----------+-----------+-------+------+
|order_id|day       |customer_id|country|amount|
+--------+----------+-----------+-------+------+
|1001    |2026-10-01|C01        |PT     |25.50 |
|1004    |2026-10-02|C01        |ES     |40.00 |
|1005    |2026-10-03|C02        |IE     |18.25 |
+--------+----------+-----------+-------+------+

It is a left join, so an order with no matching version would stay in the view with an empty country, and the second check would stop the run. Now remove the 2 lines with valid_from and valid_to from the view and run it again. It stops at the first check:

[USER_RAISED_EXCEPTION] sales has an order more than once SQLSTATE: P0001

SELECT count(*) AS rows, sum(amount) AS revenue FROM sales still returns 3 rows and 83.75: the table readers see was never replaced. Put the 2 lines back.

The comments are for the readers too: Databricks' advice for Genie, in part 3, starts with well-described tables and columns. Part 2 defines the shop's revenue once, so every reader computes it the same way.