Realtime Compute for Apache Flink:Primary key tables and append-only tables

If you specify 'merge-engine' = 'first-row', Paimon retains only the first data record in cases where multiple records have the same primary keys. Compared with the deduplicate engine, the first-row engine only generates INSERT-type changelogs, which increases the efficiency of changelog production.

In this example, the following DDL statement is used to create a Paimon table.

CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'first-row'
);

If +I(1, 2.0, 'apple'), +I(1, 4.0, 'banana'), and +I(1, 8.0, 'cherry') are written to the table in sequence, the result of the SELECT * FROM T WHERE k = 1 query is (1, 2.0, 'apple').

Paimon aggregates each column in data records that have the same primary keys based on the specified aggregate function. Use fields.<field-name>.aggregate-function to specify an aggregate function for each column that is not included in the primary keys. Otherwise, the last_non_null_value aggregate function is used.

In the following example, the price column is aggregated by using the max function, and the sales column is aggregated by using the sum function.

CREATE TABLE T (
  product_id BIGINT,
  price DOUBLE,
  sales BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'merge-engine' = 'aggregation',
  'fields.price.aggregate-function' = 'max',
  'fields.sales.aggregate-function' = 'sum'
);

If +I(1, 23.0, 15) and +I(1, 30.2, 20) are written to the table in sequence, the result of the SELECT * FROM T WHERE product_id = 1 query is (1, 30.2, 35).

The following list describes the supported aggregate functions and data types:

• sum: supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE.

• product: supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE.

• count: supports INTEGER and BIGINT.

• max or min: supports CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ.

• first_value or last_value: supports all data types, including null.

• first_not_null_value and last_non_null_value: supports all data types.

• listagg: supports STRING.

• bool_and and bool_or: BOOLEAN

If you specify 'merge-engine' = 'partial-update', you can gradually update the columns in the existing data record by using the most recent values from records that have the same primary keys. Null values do not overwrite existing values.

In this example, the following DDL statement is used to create a Paimon table.

CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 BIGINT,
  v3 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update'
);

If +I(1, 23.0, 10, NULL), +I(1, NULL, NULL, 'This is a book'), and +I(1, 25.2, NULL, NULL) are written to the table in sequence, the result of the SELECT * FROM T WHERE k = 1 query is (1, 25.2, 10, 'This is a book').

You can also specify parameters related to sequence groups and aggregate functions in the WITH clause.

• Use sequence groups to specify the merge order

  If you join columns from different tables to create a wide table, you can use sequence groups to specify the partial-update order of different columns to handle out-of-order data records.

  In the following example, columns a and b are updated in ascending order of the values in column g_1, and columns c and d are updated in ascending order of the values in column g_2.

  CREATE TABLE T (
    k INT,
    a STRING,
    b STRING,
    g_1 INT,
    c STRING,
    d STRING,
    g_2 INT,
    PRIMARY KEY (k) NOT ENFORCED
  ) WITH (
    'merge-engine' = 'partial-update',
    'fields.g_1.sequence-group' = 'a,b',
    'fields.g_2.sequence-group' = 'c,d'
  );

• Use sequence groups and aggregate functions together

  You can also specify fields.<field-name>.aggregate-function in the WITH clause to apply aggregate functions to the column specified by <field-name>. The column specified by <field-name> must belong to a sequence group. You can use all aggregate functions supported by the aggregation engine.

  In the following example, columns a and b are updated in ascending order of the values in column g_1. The final result contains the most recent non-null value in column a and the maximum value in column b. Columns c and d are updated in ascending order of the values in column g_2. The final result contains the most recent non-null value in column c and the sum of values in column b.

  CREATE TABLE T (
    k INT,
    a STRING,
    b INT,
    g_1 INT,
    c STRING,
    d INT,
    g_2 INT,
    PRIMARY KEY (k) NOT ENFORCED
  ) WITH (
    'merge-engine' = 'partial-update',
    'fields.g_1.sequence-group' = 'a,b',
    'fields.b.aggregate-function' = 'max',
    'fields.g_2.sequence-group' = 'c,d',
    'fields.d.aggregate-function' = 'sum'
  );