Realtime Compute for Apache Flink:Create and use materialized tables

Limitations

• You can modify the query only for materialized tables created in VVR 11.1 or later.

• When modifying a query, you can only add columns and modify computation logic. You cannot change the order of existing columns or modify their definitions.

  Operation	Supported	Description
  Add a new column	Yes	You can append new columns to the schema while maintaining the existing column order.
  Modify computation logic for an existing column (without changing its name or type)	Yes	You can modify the computation logic, but the column name and data type must remain the same.
  Change the order of existing columns	No	The column order is fixed. To change it, you must drop and re-create the materialized table.
  Modify the name or data type of an existing column	No	You must drop and re-create the materialized table.

Example

1. Click Edit Table and modify the query. The following code provides an example:

   ALTER MATERIALIZED TABLE `paimon`.`default`.`mt-orders`
       AS
       SELECT
         *,
         price * quantity AS total_price
       FROM orders
       WHERE price * quantity > 1000
   ;

2. Click Preview to see a before-and-after comparison.

   

3. Click OK. You can view the newly added column and modified query logic on the Table Schema tab.

Incremental updates

CREATE MATERIALIZED TABLE mt_shipped_orders (
    PRIMARY KEY (order_id) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
SELECT 
    order_id,
    COALESCE(customer_id, 'Unknown') AS customer_id,
    CAST(order_amount AS DECIMAL(10, 2)) AS order_amount,
    CASE 
        WHEN status = 'shipped' THEN 'Completed'
        WHEN status = 'pending' THEN 'In Progress'
        ELSE 'Unknown'
    END AS order_status,
    DATE_FORMAT(order_ts, 'yyyyMMdd') AS order_date,
    UDSF_ProcessFunction(notes) AS notes
FROM 
    orders
WHERE
    status = 'shipped';

CREATE MATERIALIZED TABLE mt_enriched_orders (
    PRIMARY KEY (order_id, order_tag) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
WITH o AS (
    SELECT
        order_id,
        product_id,
        quantity,
        proc_time,
        e.tag AS order_tag
    FROM 
        orders,
        LATERAL TABLE(UDTF_StringSplitFunction(tags, ',')) AS e(tag))
SELECT 
    o.order_id,
    o.product_id,
    p.product_name,
    p.category,
    o.quantity,
    p.price,
    o.quantity * p.price AS total_amount,
    order_tag
FROM o 
LEFT JOIN 
    product_info FOR SYSTEM_TIME AS OF PROCTIME() AS p
ON 
    o.product_id = p.product_id;

Data backfilling

Previously, correcting stream processing results for historical data required developing a separate batch job. By using materialized tables, you can directly backfill historical data partitions. This approach unifies batch and streaming processing, which reduces development and O&M costs.

1. Click the target materialized table under its catalog.

2. On the Data Information tab, backfill the data.

   If you defined a partitioning column when you created the materialized table, it is a partitioned table. Otherwise, it is a non-partitioned table.

   Partitioned table

   In the Data Partitions section, click Trigger Update if this is the first time you are backfilling data or if the required partition does not exist. If partitions already exist, you can select a specific partition and click Refresh in the Actions column.

   

   

   Parameters

   • Partitioning column: The partitioning column of the table. For example, if you enter 20241201, all data in the partition ds=20241201 is backfilled.

   • Task Name: The name of the data backfilling task.

   • Update Range (Optional): Specifies whether to cascade updates to downstream materialized tables. Starting from the current table, all materialized tables in the data lineage are updated. The maximum supported depth of the downstream lineage is 6 levels.

     Note

     • When updating a partitioned table, the downstream materialized tables must have the exact same partitioning columns as the starting table. Otherwise, the update operation fails.

     • If an update fails for any materialized table in the lineage, all subsequent downstream nodes will also fail.

   • Deployment Target: You can select a queue or a session cluster. The default selection is default-queue.

   Non-partitioned table

   In the Data Status section, click Refresh.

   

   Parameters

   • Task Name: The name of the data backfilling task.

   • Update Range: This option is not available for non-partitioned tables.

     Note

     • During an update, downstream data is fully refreshed.

     • If an update fails for any materialized table in the lineage, all subsequent downstream nodes will also fail.

     • Cascading updates are not supported if the starting table is a non-partitioned table that is updated by a streaming job.

   • Deployment Target: You can select a queue or a session cluster. The default selection is default-queue.

3. Scheduled and batch backfilling.

   You can use Task Orchestration to create a workflow for the materialized table to run backfilling jobs on a schedule. You can also use the workflow's data backfilling feature to backfill data in bulk for a specified time range.

Modify data freshness

1. Under the corresponding catalog, click the materialized table database, and then click the target materialized table.

2. In the upper-right corner, click Edit Freshness.

   • If the materialized table does not have a primary key, you cannot switch its update method between streaming and batch. The system uses a streaming job for data freshness values under 30 minutes and a batch job for values of 30 minutes or more. Therefore, changing the freshness across this 30-minute threshold is not permitted for tables without a primary key.

   • If the upstream table is a materialized table, ensure that the data freshness of the downstream table is a positive integer multiple of the upstream table's data freshness.

   • The maximum data freshness is one day.

View data lineage

In the left-side navigation pane, choose Operations Center > Data Lineage to go to the data lineage page for materialized tables. On this page, you can view the lineage relationships between all materialized tables. You can also perform operations such as Start/Stop Update and Edit Freshness directly on a materialized table. Click Details to navigate to the details page of the corresponding materialized table.