Delta Query rewriting optimization

Engine version

• Only Realtime Compute for Apache Flink that uses VVR 8.0.11 or later supports Delta Join.

• VVR 8.0.11 or later supports INNER JOIN, VVR 11.3 or later supports LEFT, RIGHT, and FULL OUTER JOIN, and VVR 11.4 or later supports cascading Delta Joins.

Query structure

• Only regular join operators in streaming deployments can be rewritten. Window joins, interval joins, temporal joins, and lookup joins cannot be rewritten.

• The join condition must contain at least one equality condition, and the columns covered by the equality conditions must match a single index of the source table on both sides.

• The index columns that correspond to the join key cannot participate in any computation between the source and the join, or between cascading joins.

• Non-deterministic functions such as Rand() cannot be used between the source and the join, or between cascading joins. Otherwise, the same result cannot be reproduced during a lookup.

• The nodes allowed in a Delta Join deployment are restricted. Nodes that are not in the whitelist are not supported.

  • Only Project and Filter nodes are allowed between the source and the join, and between cascading joins. Watermarks cannot be defined on the source table. A watermark introduces a WatermarkAssigner node after the source, which is not in the current whitelist.

  • Only Project, Filter, Lookup (supported in VVR 11.6 or later), and Union (supported in VVR 11.8 or later) nodes are allowed between the join and the sink.

  • If you use BEGIN STATEMENT SET to write to multiple sinks:

    • In versions earlier than VVR 11.8, each write path must contain a join node.

    • In VVR 11.8 or later, one of the write paths can omit the join node, but that path can contain only Project, Filter, Lookup, and Union nodes.

Business logic

• Your business can tolerate that data cannot be physically deleted from the result table.

Delta Join sends only +I and +U to the downstream, and does not send -U or -D. Take the filter condition WHERE l.amount + r.amount >= 100 as an example. When the value of a join record drops from 120 to 80:

+U(..., 120) → The condition is met and the record is written to the result table.
+U(..., 80)  → The condition is not met and the record is filtered out.

Because no -U is sent to retract the previous result, the record whose amount=120 remains in the result table.

By default, a query that contains a filter condition on non-unique keys cannot be converted into a Delta Join. If your business can tolerate residual data, you can set 'table.optimizer.delta-join.ignore-non-unique-key-filter' = 'true' to skip the check.

For scenarios in which data must be deleted, we recommend that you use logical deletion and rewrite the filter condition into a flag column. In the following query, each upstream change overwrites the record in the result table, and is_delete always reflects the latest state. A downstream query only needs to add WHERE is_delete = 0.

INSERT INTO order_item_wide
SELECT
  o.merchant_id, o.order_id, o.item_id, i.item_name, o.amount,
  CASE WHEN o.amount >= 100 THEN 0 ELSE 1 END AS is_delete
FROM orders AS o
JOIN items AS i
  ON o.merchant_id = i.merchant_id AND o.item_id = i.item_id;

The same applies when a record is deleted from the source table. The delete message is not propagated to the result table, and the join results that were written earlier are retained. If the downstream must be aware of the deletion, use the preceding logical deletion solution and let the upstream express the deletion state by using a flag column.

• The relationship between the primary key and the join key must remain unchanged to avoid disorder in distributed processing.

For example, the primary key of a source table is the user_id column, and the join key on the table is user_name. After a record whose user_id is 1 and whose user_name is Jim is generated, no matter how the data whose user_id is 1 changes later, user_name cannot be changed to another value, such as Sam.

Step 1: Create the source tables and the result table

The join key must match an index on both source tables. In this example, the join key is (merchant_id, item_id). In the orders table, the join key is a prefix of the primary key and must be declared as the bucket key to serve as an available prefix index. In the items table, the join key is the primary key, so no extra configuration is required.

CREATE TABLE `my-catalog`.`my_db`.`orders` (
  merchant_id BIGINT,                    -- Merchant ID
  item_id BIGINT,                        -- Item ID
  order_id BIGINT,                       -- Order ID
  amount DECIMAL(18, 2),                 -- Current order amount
  PRIMARY KEY (merchant_id, item_id, order_id) NOT ENFORCED
) WITH (
  'bucket.key' = 'merchant_id,item_id'
);

CREATE TABLE `my-catalog`.`my_db`.`items` (
  merchant_id BIGINT,                    -- Merchant ID
  item_id BIGINT,                        -- Item ID
  item_name STRING,                      -- Item name
  PRIMARY KEY (merchant_id, item_id) NOT ENFORCED   -- The join key is the same as the primary key.
);

CREATE TABLE `my-catalog`.`my_db`.`order_item_wide` (
  merchant_id BIGINT,
  order_id BIGINT,
  item_id BIGINT,
  item_name STRING,
  amount DECIMAL(18, 2),
  PRIMARY KEY (merchant_id, order_id, item_id) NOT ENFORCED
);

Step 2: Write the join deployment

USE CATALOG `my-catalog`;
USE `my_db`;

-- Enable Delta Join rewriting.
SET 'table.optimizer.delta-join.strategy' = 'EVENTUAL';

-- A Delta Join statement does not require dedicated syntax.
INSERT INTO order_item_wide
SELECT
  o.merchant_id,
  o.order_id,
  o.item_id,
  i.item_name,
  o.amount
FROM orders AS o
JOIN items AS i
  ON o.merchant_id = i.merchant_id AND o.item_id = i.item_id;

Step 3: Verify that the optimization takes effect

After the deployment is published and started, check the deployment topology on the Status page. If you see the following Delta Join node, the streaming join is rewritten into a Delta Join.

Flink deployment parameters

Fluss table parameters

The following parameters are configured in the WITH clause when you create a Fluss table. You can also use SQL hints to adjust them for a single deployment.

When you use hints to adjust these parameters, place the hints after the table that is looked up:

INSERT INTO order_item_wide
SELECT o.merchant_id, o.order_id, o.item_id, i.item_name, o.amount
FROM orders /*+ OPTIONS('client.lookup.queue-size' = '51200') */ AS o
JOIN items /*+ OPTIONS('client.lookup.queue-size' = '51200') */ AS i
  ON o.merchant_id = i.merchant_id AND o.item_id = i.item_id;

Engine version

• Only Realtime Compute for Apache Flink that uses VVR 11.8 or later supports Delta Aggregate.

Query structure

• Query structure

  • Only single-level non-windowed group aggregations in streaming deployments can be rewritten. Window aggregations and cascading aggregations cannot be rewritten.

  • The group key must contain a single index of the source table.

  • The index columns that correspond to the group key cannot participate in any computation between the source and the aggregation.

  • Non-deterministic functions such as Rand() cannot be used between the source and the aggregation. Otherwise, the same result cannot be reproduced during a lookup.

  • The nodes allowed in a Delta Aggregate deployment are restricted. Nodes that are not in the whitelist are not supported.

    • Only Project, Filter, and MiniBatchAssigner nodes are allowed between the source and the aggregation. Watermarks cannot be defined on the source table.

    • Only Project and Filter nodes are allowed between the aggregation and the sink.

    • If you use BEGIN STATEMENT SET to write to multiple sinks, each write path must contain an aggregation node.

• Aggregate functions

  • SUM, COUNT, AVG, MIN, MAX, LISTAGG, FIRST_VALUE, and LAST_VALUE are supported.

  • DISTINCT, aggregations with a FILTER clause, UDAFs, Python aggregate functions, and other aggregate functions that are not listed are not supported.

Business logic

• Your business can tolerate that data cannot be physically deleted from the result table.

Delta Aggregate sends only +U and -D to the downstream, and does not send -U. Take the filter condition HAVING SUM(amount) >= 100 as an example. When the SUM(amount) value of a group is updated from 120 to 80:

+U(..., 120) → The condition is met and the record is written to the result table.
+U(..., 80)  → The condition is not met and the record is filtered out.

Because no -U is sent to retract the previous result, the record whose SUM(amount)=120 remains in the result table.

By default, a query that contains a filter condition on non-unique keys cannot be converted into a Delta Aggregate. If your business can tolerate residual data, you can set 'table.optimizer.delta-agg.ignore-non-unique-key-filter' = 'true' to skip the check.

For scenarios in which data must be deleted, we also recommend logical deletion. For more information about how to write the statement, see the corresponding description of Delta Join.

Unlike Delta Join, Delta Aggregate does not discard delete messages when records are deleted from the source table. It generates the correct aggregation result and sends the result to the downstream. When all detail records in a group are deleted, Delta Aggregate sends a -D message to the downstream.

• The relationship between the primary key and the group key must remain unchanged to avoid disorder in distributed processing.

For example, the primary key of a source table is the user_id column, and the group key on the table is user_name. After a record whose user_id is 1 and whose user_name is Jim is generated, no matter how the data whose user_id is 1 changes later, user_name cannot be changed to another value, such as Sam.

Step 1: Create the source table and the result table

In this example, GROUP BY merchant_id does not cover the primary key (merchant_id, order_id) of the source table, but it is a prefix of the primary key. Therefore, it is declared as the bucket key to serve as an available prefix index.

CREATE TABLE `my-catalog`.`my_db`.`orders` (
  merchant_id BIGINT,                    -- Merchant ID
  order_id BIGINT,                       -- Order ID
  amount DECIMAL(18, 2),                 -- Current order amount
  PRIMARY KEY (merchant_id, order_id) NOT ENFORCED
) WITH (
  'bucket.key' = 'merchant_id'           -- Look up all orders of a merchant by merchant_id.
);

CREATE TABLE `my-catalog`.`my_db`.`merchant_summary` (
  merchant_id BIGINT,
  order_count BIGINT,                    -- The current number of orders of the merchant.
  total_amount DECIMAL(38, 2),           -- The current total transaction amount of the merchant. The precision must be higher than that of the source table field to avoid overflow during accumulation.
  PRIMARY KEY (merchant_id) NOT ENFORCED
);

Step 2: Write the aggregation deployment

USE CATALOG `my-catalog`;
USE `my_db`;

-- Enable Delta Aggregate rewriting.
SET 'table.optimizer.delta-agg.strategy' = 'EVENTUAL';

-- A Delta Aggregate statement does not require dedicated syntax.
INSERT INTO merchant_summary
SELECT
  merchant_id,
  COUNT(*) AS order_count,
  SUM(amount) AS total_amount
FROM orders
GROUP BY merchant_id;

Step 3: Verify that the optimization takes effect

After the deployment is published and started, check the deployment topology on the Status page. If you see the following Delta Aggregate node, the aggregation is rewritten into a Delta Aggregate.

Flink deployment parameters

Fluss table parameters

Lookups of Delta Aggregate also go through the Fluss client. The configurable parameters and the tuning suggestions are exactly the same as those of Delta Join. For more information, see the Fluss table parameters section of Delta Join.