Realtime Compute for Apache Flink:Create and use materialized tables

Limits

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

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

  Operation Type	Support	Description
  Append a new column	Support	You can add a new column to the end of the schema while keeping the existing column order.
  Modify the calculation logic of an existing column (without changing the column name and type)	Support	For example, you can modify the calculation logic, but the column name and data type must remain the same.
  Change the order of existing columns	No	The column order is fixed. You must delete and re-create the materialized table.
  Modify the name or data type of an existing column	No	You must delete and re-create the materialized table.

Modification 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 view the changes.

   

3. Click Confirm. You can then view the new column and query logic on the Table Schema Details tab.

Incremental update of a materialized table

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;

Backfill historical data

Previously, after running a stream job, you had to develop a separate batch job to reconcile the results with the full dataset from the previous day. With materialized tables, you can directly select a historical data partition to backfill data. This change reduces development and O&M costs and provides unified stream and batch processing.

1. Click the target materialized table under the catalog.

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

   If you declared partition fields when you created the materialized table, it is a partitioned table. Otherwise, it is a non-partitioned table.

   Partitioned table

   View the Data Partitions. If this is the first backfill or the required partition does not exist, click Manual Update. If partitions exist, select the partition to backfill and click Backfill.

   

   

   Parameters:

   • Partition Field: The partition field of the table. For example, if you enter `20241201`, all data where `ds=20241201` is backfilled.

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

   • Update Scope (Optional): Specifies whether to cascade updates to associated downstream materialized tables. The update cascades from the current table through all materialized tables in its data lineage. The maximum number of downstream layers is 6.

     Note

     • For partitioned table updates, the partition fields of the downstream materialized tables must be identical to those of the starting table. Otherwise, the update operation fails.

     • If a materialized table in the lineage fails to update, all downstream nodes also fail.

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

   Non-partitioned table

   View the Data Status and click Backfill.

   

   Parameters:

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

   • Update Scope: This parameter is not available for non-partitioned tables.

     Note

     • During the update, a full refresh is performed on downstream data.

     • If a materialized table in the lineage fails to update, all downstream nodes also fail.

     • If the system determines that the starting table is a non-partitioned stream task based on its freshness, cascade updates are not supported.

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

3. Scheduled and batch backfills.

   You can use Task Orchestration to create a materialized table workflow for recurring schedules. This enables scheduled backfills. You can also use the data backfill feature of the workflow to select a time range for batch data backfills.

Modify data freshness

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

2. In the upper-right corner, click Modify Data Freshness.

   • If the materialized table does not have a primary key, you cannot change the job's processing mode between stream and batch. For example, you cannot change the data freshness from 2 seconds (stream) to 1 hour (batch), or vice versa. A job is a stream job if its freshness is less than 30 minutes, and a batch job if its freshness is 30 minutes or more.

   • If the base table is a materialized table, the data freshness of the downstream table must be an integer multiple of the upstream table's data freshness.

   • The data freshness cannot exceed one day.

View data lineage

The data lineage page displays the lineage relationships between all materialized tables. From this page, you can also perform operations such as Start/Stop Update and Modify Data Freshness on the materialized tables. To view the details page for a specific materialized table, click Details.