Realtime Compute for Apache Flink:プライマリキーテーブルと追加専用テーブル

'merge-engine' = 'first-row' を指定すると、複数のレコードのプライマリキーが同じ場合、Paimon は最初のデータレコードのみを保持します。 deduplicate エンジンと比較して、first-row エンジンは INSERT タイプの変更ログのみを生成するため、変更ログ生成の効率が向上します。

この例では、次の DDL ステートメントを使用して Paimon テーブルを作成します。

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

+I(1, 2.0, 'apple')、+I(1, 4.0, 'banana')、+I(1, 8.0, 'cherry') がテーブルに順番に書き込まれた場合、SELECT * FROM T WHERE k = 1 クエリの結果は (1, 2.0, 'apple') になります。

Paimon は、指定された集計関数に基づいて、同じプライマリキーを持つデータレコードの各列を集計します。プライマリキーに含まれていない各列の集計関数を指定するには、fields.<field-name>.aggregate-function を使用します。それ以外の場合は、last_non_null_value 集計関数が使用されます。

次の例では、price 列は max 関数を使用して集計され、sales 列は sum 関数を使用して集計されます。

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'
);

+I(1, 23.0, 15) と +I(1, 30.2, 20) がテーブルに順番に書き込まれた場合、SELECT * FROM T WHERE product_id = 1 クエリの結果は (1, 30.2, 35) になります。

次のリストは、サポートされている集計関数とデータ型を示しています。

• sum：DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE をサポートします。

• product：DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE をサポートします。

• count：INTEGER と BIGINT をサポートします。

• max または min：CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ をサポートします。

• first_value または last_value：null を含むすべてのデータ型をサポートします。

• first_not_null_value と last_non_null_value：すべてのデータ型をサポートします。

• listagg：STRING をサポートします。

• bool_and と bool_or：BOOLEAN

'merge-engine' = 'partial-update' を指定すると、同じプライマリキーを持つレコードの最新値を使用して、既存のデータレコードの列を徐々に更新できます。 null 値は既存の値を上書きしません。

この例では、次の DDL ステートメントを使用して Paimon テーブルを作成します。

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

+I(1, 23.0, 10, NULL)、+I(1, NULL, NULL, 'This is a book')、+I(1, 25.2, NULL, NULL) がテーブルに順番に書き込まれた場合、SELECT * FROM T WHERE k = 1 クエリの結果は (1, 25.2, 10, 'This is a book') になります。

WITH 句でシーケンスグループと集計関数に関連するパラメーターを指定することもできます。

• シーケンスグループを使用してマージ順序を指定する

  異なるテーブルの列を結合してワイドテーブルを作成する場合、シーケンスグループを使用して異なる列の partial-update 順序を指定し、順序が正しくないデータレコードを処理できます。

  次の例では、列 a と b は列 g_1 の値の昇順で更新され、列 c と d は列 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'
  );

• シーケンスグループと集計関数を一緒に使用する

  WITH 句で fields.<field-name>.aggregate-function を指定して、<field-name> で指定された列に集計関数を適用することもできます。<field-name> で指定された列は、シーケンスグループに属している必要があります。 aggregation エンジンでサポートされているすべての集計関数を使用できます。

  次の例では、列 a と b は列 g_1 の値の昇順で更新されます。最終結果には、列 a の最新の null 以外の値と列 b の最大値が含まれます。列 c と d は列 g_2 の値の昇順で更新されます。最終結果には、列 c の最新の null 以外の値と列 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'
  );