Apache Paimon — primary-key tables, merge engines, partial-update

6 h
requires STH-01

From your world. MDM golden record: one customer row assembled from three source systems

Build a customer golden record from 3 streams with Paimon partial-update

Three source systems each know part of a customer:

CRM knows name
billing knows email
loyalty knows phone

Each system writes a row keyed by customer_id. It fills only its own column and leaves the others NULL. Your job is to build the Paimon table and the insert statements for each source. The three partial rows must merge into one complete golden record for each customer. This is continuous MDM in the storage layer.

Implement, in solution.py:

def table_ddl(table: str) -> str:
    """Return the CREATE TABLE statement for the golden-record table `table`
    with columns (customer_id INT, name STRING, email STRING, phone STRING)."""

def source_inserts(table: str, crm, billing, loyalty) -> list[str]:
    """Return INSERT statements that write each source's rows into `table`,
    filling only that source's column(s) and NULL for the rest."""

The inputs to source_inserts are lists of tuples:

crm: [(customer_id, name), ...]
billing: [(customer_id, email), ...]
loyalty: [(customer_id, phone), ...]

Requirements

The table must use the partial-update merge engine, a PRIMARY KEY(customer_id) NOT ENFORCED, and 'bucket'='1' (deterministic layout).
Each INSERT fills only the column(s) of the source that owns them. The other string columns are CAST(NULL AS STRING).
customer_id is an INT. name, email and phone are STRING.

What the tests do (G6 oracle)

The test runs your DDL and inserts through the native embedded Paimon runner (paimon_sql.batch, no Docker). Then it reads SELECT * FROM <table> ORDER BY customer_id. It compares the merged rows with an independent Python oracle. The oracle computes the golden record in a different way (latest non-null value per column). The test also checks these points:

A customer known to all three sources gets all three fields (the merge really combines them).
A later NULL from one source does not erase a value that another source set.
The result is the same whether a customer appears in one, two or three sources.

The whole suite may only skip. This happens when the vendored Flink/JDK/Paimon jars are missing. Then run bash labs/fetch_streamhouse.sh first. The fixture prints the exact command.

Starter code

solution.py

"""STH-02/01 — build a customer golden record from 3 streams with Paimon.

TODO: pick the right merge engine so partial rows from CRM, billing and loyalty
COMBINE into one complete record. The starter below uses the DEFAULT engine,
which keeps the last whole record and loses the other sources' columns.
"""


def table_ddl(table: str) -> str:
    # BUG: no merge-engine set -> defaults to 'deduplicate' (last-writer-wins),
    # so each source's NULLs overwrite the others. Use 'partial-update'.
    return (
        f"CREATE TABLE {table} ("
        "customer_id INT, name STRING, email STRING, phone STRING, "
        "PRIMARY KEY(customer_id) NOT ENFORCED) "
        "WITH ('bucket'='1')"
    )


def _lit(v):
    return "CAST(NULL AS STRING)" if v is None else "'" + str(v).replace("'", "''") + "'"


def source_inserts(table, crm, billing, loyalty):
    stmts = []
    for cid, name in crm:
        stmts.append(
            f"INSERT INTO {table} VALUES "
            f"({cid}, {_lit(name)}, CAST(NULL AS STRING), CAST(NULL AS STRING))"
        )
    for cid, email in billing:
        stmts.append(
            f"INSERT INTO {table} VALUES "
            f"({cid}, CAST(NULL AS STRING), {_lit(email)}, CAST(NULL AS STRING))"
        )
    for cid, phone in loyalty:
        stmts.append(
            f"INSERT INTO {table} VALUES "
            f"({cid}, CAST(NULL AS STRING), CAST(NULL AS STRING), {_lit(phone)})"
        )
    return stmts

Hints

Hint 1

A golden record assembled from partial source rows needs the partial-update merge engine, not the default deduplicate. Declare PRIMARY KEY(customer_id) NOT ENFORCED and set 'merge-engine'='partial-update'.

Hint 2

Pin 'bucket'='1' so the small table has a deterministic layout. Each source writes a row keyed by customer_id with only its own column(s) filled and the rest CAST(NULL AS STRING).

Hint 3

partial-update keeps the latest NON-NULL value per column, so a later NULL never erases an earlier value. The three partial rows for one customer merge into one complete row; your DDL + the per-source inserts are what the test runs.

Run it locally

uv run de start STH-02/01-golden-record
# edit workspace/STH-02/01-golden-record/
uv run de test