E-MapReduce:Accelerate data lake queries using materialized views

Choose a suitable refresh policy

Currently, StarRocks cannot detect partition-level data changes in Hudi Catalogs. Therefore, once a refresh task is triggered, a full refresh is performed.

For Hive Catalogs, Iceberg Catalogs (from v3.1.4), JDBC Catalogs (from v3.1.4, and only MySQL Range partitions are supported), and Paimon Catalogs (from v3.2.1), StarRocks supports detecting partition-level data changes. As a result, StarRocks can:

• Refresh only the partitions where data has changed. This avoids full refreshes and reduces the resource consumption caused by the refresh process.

• Ensure data consistency to some extent during query rewrite. If the base table in the data lake undergoes data changes, the query will not be rewritten to use the materialized view.

Note that for partition-based refreshes, the partition key of the materialized view must be included in the partition key of the base table.

From v3.2.3, StarRocks supports creating partitioned materialized views on Iceberg tables that use Partition Transforms. The materialized view will be partitioned based on the transformed columns. Currently, only Iceberg tables that use the identity, year, month, day, or hour Transform are supported.

The following example shows the definition of an Iceberg table that uses the day Transform and a partition-aligned materialized view created on that table:

-- Definition of the Iceberg table.
CREATE TABLE spark_catalog.test_db.iceberg_sample_datetime_day (
  id         BIGINT,
  data       STRING,
  category   STRING,
  ts         TIMESTAMP)
USING iceberg
PARTITIONED BY (days(ts))

-- Create a materialized view based on the preceding Iceberg table.
CREATE MATERIALIZED VIEW `test_iceberg_datetime_day_mv` (`id`, `data`, `category`, `ts`)
PARTITION BY (`ts`)
DISTRIBUTED BY HASH(`id`)
REFRESH MANUAL
AS 
SELECT 
  `iceberg_sample_datetime_day`.`id`, 
  `iceberg_sample_datetime_day`.`data`, 
  `iceberg_sample_datetime_day`.`category`, 
  `iceberg_sample_datetime_day`.`ts`
FROM `iceberg`.`test`.`iceberg_sample_datetime_day`;

For Hive Catalogs, you can enable the Hive metadata cache refresh feature to allow StarRocks to detect data changes at the partition level. After this feature is enabled, StarRocks periodically accesses the Hive Metastore service (HMS) or AWS Glue to check the metadata information of frequently queried hot data.

Configuration items

To enable the Hive metadata cache refresh feature, you can use ADMIN SET FRONTEND CONFIG to set the following FE dynamic configuration items. The syntax is as follows.

ADMIN SET FRONTEND CONFIG ("key" = "value")

For Iceberg Catalogs, from v3.1.4, StarRocks supports detecting partition-level data changes. Currently, only Iceberg V1 tables are supported.

Enable query rewrite for External Catalog materialized views

Because strong data consistency cannot be guaranteed, StarRocks disables the query rewrite feature for Hudi and JDBC Catalog materialized views by default. You can enable this feature by setting the property force_external_table_query_rewrite to true when you create the materialized view. For materialized views created on tables in a Hive Catalog, the query rewrite feature is enabled by default. In scenarios that involve query rewrite, if you use a very complex query statement to build the materialized view, we recommend that you split the query statement and build multiple simple materialized views in a nested manner. Nested materialized views are more flexible and can adapt to a wider range of query patterns.

The following is an example.

CREATE MATERIALIZED VIEW ex_mv_par_tbl
PARTITION BY emp_date
DISTRIBUTED BY hash(empid)
PROPERTIES (
"force_external_table_query_rewrite" = "true"
) 
AS
SELECT empid, deptno, emp_date
FROM `hudi_catalog`.`emp_db`.`emps_par_tbl`
WHERE empid < 5;

Case 1: Accelerate Join computations in the data lake

You can use materialized views to accelerate Join queries in the data lake.

Assume the following queries on a Hive catalog are slow queries.

--Q1
SELECT SUM(lo_extendedprice * lo_discount) AS REVENUE
FROM hive.ssb_1g_csv.lineorder, hive.ssb_1g_csv.dates
WHERE
    lo_orderdate = d_datekey
    AND d_year = 1993
    AND lo_discount BETWEEN 1 AND 3
    AND lo_quantity < 25;

--Q2
SELECT SUM(lo_extendedprice * lo_discount) AS REVENUE
FROM hive.ssb_1g_csv.lineorder, hive.ssb_1g_csv.dates
WHERE
    lo_orderdate = d_datekey
    AND d_yearmonth = 'Jan1994'
    AND lo_discount BETWEEN 4 AND 6
    AND lo_quantity BETWEEN 26 AND 35;

--Q3 
SELECT SUM(lo_revenue), d_year, p_brand
FROM hive.ssb_1g_csv.lineorder, hive.ssb_1g_csv.dates, hive.ssb_1g_csv.part, hive.ssb_1g_csv.supplier
WHERE
    lo_orderdate = d_datekey
    AND lo_partkey = p_partkey
    AND lo_suppkey = s_suppkey
    AND p_brand BETWEEN 'MFGR#2221' AND 'MFGR#2228'
    AND s_region = 'ASIA'
GROUP BY d_year, p_brand
ORDER BY d_year, p_brand;

By analyzing the query profiles, you might notice that most of the query execution time is spent on the Hash Join between the table lineorder and other dimension tables on the column lo_orderdate.

Here, Q1 and Q2 perform aggregation after joining lineorder and dates, while Q3 performs aggregation after joining lineorder, dates, part, and supplier.

Therefore, you can leverage StarRocks' View Delta Join rewrite capability to build a materialized view that joins lineorder, dates, part, and supplier.

CREATE MATERIALIZED VIEW lineorder_flat_mv
DISTRIBUTED BY HASH(LO_ORDERDATE, LO_ORDERKEY) BUCKETS 48
PARTITION BY LO_ORDERDATE
REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
PROPERTIES (
    -- Specify the UNIQUE constraint.
    "unique_constraints" = "
    hive.ssb_1g_csv.supplier.s_suppkey;
    hive.ssb_1g_csv.part.p_partkey;
    hive.ssb_1g_csv.dates.d_datekey",
    -- Specify the foreign key constraint.
    "foreign_key_constraints" = "
    hive.ssb_1g_csv.lineorder(lo_partkey) REFERENCES hive.ssb_1g_csv.part(p_partkey);
    hive.ssb_1g_csv.lineorder(lo_suppkey) REFERENCES hive.ssb_1g_csv.supplier(s_suppkey);
    hive.ssb_1g_csv.lineorder(lo_orderdate) REFERENCES hive.ssb_1g_csv.dates(d_datekey)",
    -- Enable query rewrite.
    "force_external_table_query_rewrite" = "TRUE"
)
AS SELECT
       l.LO_ORDERDATE AS LO_ORDERDATE,
       l.LO_ORDERKEY AS LO_ORDERKEY,
       l.LO_PARTKEY AS LO_PARTKEY,
       l.LO_SUPPKEY AS LO_SUPPKEY,
       l.LO_QUANTITY AS LO_QUANTITY,
       l.LO_EXTENDEDPRICE AS LO_EXTENDEDPRICE,
       l.LO_DISCOUNT AS LO_DISCOUNT,
       l.LO_REVENUE AS LO_REVENUE,
       s.S_REGION AS S_REGION,
       p.P_BRAND AS P_BRAND,
       d.D_YEAR AS D_YEAR,
       d.D_YEARMONTH AS D_YEARMONTH
   FROM hive.ssb_1g_csv.lineorder AS l
            INNER JOIN hive.ssb_1g_csv.supplier AS s ON s.S_SUPPKEY = l.LO_SUPPKEY
            INNER JOIN hive.ssb_1g_csv.part AS p ON p.P_PARTKEY = l.LO_PARTKEY
            INNER JOIN hive.ssb_1g_csv.dates AS d ON l.LO_ORDERDATE = d.D_DATEKEY;

Case 2: Accelerate aggregations and post-join aggregations in the data lake

Materialized views can be used to accelerate aggregate queries, whether on a single table or involving multiple tables.

• Single-table aggregate query

  For a typical single-table query, if the Query Profile shows that the AGGREGATE node consumes a significant amount of time, you can build a materialized view that uses common aggregate operators. Assume the following query is slow.

  --Q4
  SELECT
  lo_orderdate, count(distinct lo_orderkey)
  FROM hive.ssb_1g_csv.lineorder
  GROUP BY lo_orderdate
  ORDER BY lo_orderdate limit 100;

  Q4 is a query that calculates the number of distinct daily orders. Because count distinct is resource-intensive, you can create the following two types of materialized views:

  CREATE MATERIALIZED VIEW mv_2_1 
  DISTRIBUTED BY HASH(lo_orderdate)
  PARTITION BY LO_ORDERDATE
  REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
  AS 
  SELECT
  lo_orderdate, count(distinct lo_orderkey)
  FROM hive.ssb_1g_csv.lineorder
  GROUP BY lo_orderdate;
  
  CREATE MATERIALIZED VIEW mv_2_2 
  DISTRIBUTED BY HASH(lo_orderdate)
  PARTITION BY LO_ORDERDATE
  REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
  AS 
  SELECT
  -- lo_orderkey must be of BIGINT type to be used for query rewrite.
  lo_orderdate, bitmap_union(to_bitmap(lo_orderkey))
  FROM hive.ssb_1g_csv.lineorder
  GROUP BY lo_orderdate;

  Note

  Do not create materialized views with LIMIT and ORDER BY clauses here to avoid rewrite failures.

• Multi-table aggregate query

  In scenarios that involve aggregating join results, you can create a nested materialized view on an existing materialized view that joins multiple tables to further aggregate the results. For example, based on the example in Case 1, you can create the following materialized view to accelerate Q1 and Q2 because their aggregation patterns are similar.

  CREATE MATERIALIZED VIEW mv_2_3
  DISTRIBUTED BY HASH(lo_orderdate)
  PARTITION BY LO_ORDERDATE
  REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
  AS 
  SELECT
  lo_orderdate, lo_discount, lo_quantity, d_year, d_yearmonth, SUM(lo_extendedprice * lo_discount) AS REVENUE
  FROM lineorder_flat_mv
  GROUP BY lo_orderdate, lo_discount, lo_quantity, d_year, d_yearmonth;

  You can also perform both joins and aggregations in a single materialized view. These types of materialized views have fewer opportunities for query rewrites because the calculations are more specific. However, they use less storage space after aggregation. You can choose the best approach for your specific scenario.

  CREATE MATERIALIZED VIEW mv_2_4
  DISTRIBUTED BY HASH(lo_orderdate)
  PARTITION BY LO_ORDERDATE
  REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
  PROPERTIES (
      "force_external_table_query_rewrite" = "TRUE"
  )
  AS
  SELECT lo_orderdate, lo_discount, lo_quantity, d_year, d_yearmonth, SUM(lo_extendedprice * lo_discount) AS REVENUE
  FROM hive.ssb_1g_csv.lineorder, hive.ssb_1g_csv.dates
  WHERE lo_orderdate = d_datekey
  GROUP BY lo_orderdate, lo_discount, lo_quantity, d_year, d_yearmonth;

Case 3: Accelerate post-aggregation Join computations in the data lake

In some cases, you may need to first perform an aggregation computation on one table and then perform a Join query with other tables. To fully leverage StarRocks' query rewrite feature, we recommend that you build nested materialized views. The following is an example.

--Q5
SELECT * FROM  (
    SELECT 
      l.lo_orderkey, l.lo_orderdate, c.c_custkey, c_region, sum(l.lo_revenue)
    FROM 
      hive.ssb_1g_csv.lineorder l 
      INNER JOIN (
        SELECT distinct c_custkey, c_region 
        from 
          hive.ssb_1g_csv.customer 
        WHERE 
          c_region IN ('ASIA', 'AMERICA') 
      ) c ON l.lo_custkey = c.c_custkey
      GROUP BY  l.lo_orderkey, l.lo_orderdate, c.c_custkey, c_region
  ) c1 
WHERE 
  lo_orderdate = '19970503'

Q5 first performs an aggregation on the customer table, and then performs a Join and aggregation on the lineorder table. Similar queries may involve different filter conditions on c_region and lo_orderdate. To leverage the query rewrite feature, you can create two materialized views, one for aggregation and another for the join.

--mv_3_1
CREATE MATERIALIZED VIEW mv_3_1
DISTRIBUTED BY HASH(c_custkey)
REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
PROPERTIES (
    "force_external_table_query_rewrite" = "TRUE"
)
AS
SELECT distinct c_custkey, c_region from hive.ssb_1g_csv.customer; 

--mv_3_2
CREATE MATERIALIZED VIEW mv_3_2
DISTRIBUTED BY HASH(lo_orderdate)
PARTITION BY LO_ORDERDATE
REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
PROPERTIES (
    "force_external_table_query_rewrite" = "TRUE"
)
AS
SELECT l.lo_orderdate, l.lo_orderkey, mv.c_custkey, mv.c_region, sum(l.lo_revenue)
FROM hive.ssb_1g_csv.lineorder l 
INNER JOIN mv_3_1 mv
ON l.lo_custkey = mv.c_custkey
GROUP BY l.lo_orderkey, l.lo_orderdate, mv.c_custkey, mv.c_region;

Case 4: Hot and cold data separation for real-time and historical data in the data lake

Consider a scenario where new data from the last three days is written directly to StarRocks, while data older than three days is batch-written to Hive. However, queries may still need to access data from the past seven days. In this case, you can use a materialized view to create a simple data expiration model.

CREATE MATERIALIZED VIEW mv_4_1 
DISTRIBUTED BY HASH(lo_orderdate)
PARTITION BY LO_ORDERDATE
REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
AS 
SELECT lo_orderkey, lo_orderdate, lo_revenue
FROM hive.ssb_1g_csv.lineorder
WHERE lo_orderdate<=current_date()
AND lo_orderdate>=date_add(current_date(), INTERVAL -4 DAY);

You can further build views or materialized views based on the upper-layer business logic to encapsulate computations.