Data Lake Formation:Paimon primary key tables

Data distribution

By default, each record is assigned to a bucket based on the hash of its primary key.

Data visibility

Data visibility for postponed-bucketing tables depends on the write method. DLF enables batch write to fixed bucket when the following conditions are met, making data visible immediately after commit:

• Deletion vectors are not enabled, and data is written via Flink batch jobs or Spark jobs using INSERT INTO or INSERT OVERWRITE.

• Deletion vectors are enabled, and data is written via Flink batch jobs or Spark jobs using INSERT OVERWRITE.

In all other cases, data becomes visible after DLF processes it, typically within 1–3 minutes. If DLF is performing resource or bucket count adjustments, visibility may take 5–10 minutes. You can select a specific storage optimization mode to reduce adjustment frequency. See Storage optimization modes for details.

Data consistency

• Single-job writes: For data written by the same job, DLF processes records in write order, providing sequential consistency.

  Note

  A job restarted from a checkpoint or savepoint with unchanged parallelism is treated as the same job.

• Multiple-job writes: For data written by different jobs, the final table state may contain a mix of both jobs’ results, but no data is lost, providing snapshot isolation consistency. To control merge order, configure a sequence field. See Handling out-of-order data.

Data distribution

Same as postponed bucketing. See Data distribution.

Data visibility

Visibility depends on whether deletion vectors are enabled:

• Without deletion vectors: data is visible immediately after commit.

• With deletion vectors: data becomes visible after the DLF optimization job processes it, typically within 1–3 minutes.

Data distribution

Dynamic-bucketing tables write data to existing buckets and automatically create new buckets when a bucket exceeds its capacity. The following properties control behavior:

When dynamic bucket is configured, update performance and resource usage depend on whether the primary key fully includes the partition key:

• Primary key contains the partition key: Updates remain within the same partition and use heap memory.

• Primary key does not fully contain the partition key: Updates may cross partitions, require RocksDB, and can degrade performance.

Data visibility

Same as fixed bucket. See Data visibility.

Supported aggregate functions

• sum: Computes the sum. Types: DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE.

• product: Computes the product. Types: DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE.

• max: Returns the maximum. Types: CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ.

• min: Returns the minimum. Same types as max.

• first_value: Returns the first input value (including null). All types.

• first_not_null_value: Returns the first non-null input value. All types.

• last_value: Returns the latest input value (including null). All types.

• last_non_null_value (default): Returns the latest non-null input value. All types.

• listagg: Concatenates strings with commas. Type: STRING.

• bool_and / bool_or: Logical AND / OR. Type: BOOLEAN.

• rbm32 / rbm64: Merges 32-bit / 64-bit RoaringBitmaps. Type: VARBINARY.

• nested_update: Aggregates rows into an ARRAY. Set 'fields.<field-name>.nested-key' = 'pk0,pk1,...' to deduplicate by nested key (keeping the last entry).

• nested_partial_update: Like nested_update but performs partial column updates within the ARRAY by nested key.

• collect: Merges multiple ARRAYs into one. Set 'fields.<field-name>.distinct' = 'true' to deduplicate.

• merge_map: Merges multiple MAPs, keeping the last value for duplicate keys.

Example

Use nested_update to aggregate multiple sub orders:

-- Main orders table
CREATE TABLE orders (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_name STRING,
  address STRING
);

-- Sub order table
CREATE TABLE sub_orders (
  order_id BIGINT,
  sub_order_id INT,
  product_name STRING,
  price BIGINT,
  PRIMARY KEY (order_id, sub_order_id) NOT ENFORCED
);

-- Wide table
-- Aggregate sub orders
-- And deduplicate sub orders based on sub_order_id
CREATE TABLE order_wide (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_name STRING,
  address STRING,
  sub_orders ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>
) WITH (
  'merge-engine' = 'aggregation',
  'fields.sub_orders.aggregate-function' = 'nested_update',
  'fields.sub_orders.nested-key' = 'sub_order_id'
);

INSERT INTO order_wide

SELECT
  order_id,
  user_name,
  address,
  CAST (NULL AS ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>)
FROM orders

UNION ALL

SELECT
  order_id,
  CAST (NULL AS STRING),
  CAST (NULL AS STRING),
  ARRAY[ROW(sub_order_id, product_name, price)]
FROM sub_orders;

Assume the orders table has data:

• (1, 'Alice', 'add1'),

• (2, 'Bob', 'add2')

The sub_orders table has data:

• (1, 11, 'apple', 10),

• (1, 12, 'banana', 20),

• (2, 21, 'cherry', 30),

• (2, 22, 'peach', 40),

• (2, 21, 'cherry', 50)

The order_wide table will have data:

• (1, 'Alice', 'add1', [(11, 'apple', 10), (12, 'banana', 20)])

• (2, 'Bob', 'add2', [(21, 'cherry', 50), (22, 'peach', 40)])

Sequence groups

In addition to the global sequence field, you can use sequence groups to specify independent merge ordering for different column groups. This is useful for table-widening scenarios with multiple source tables.

Example:

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

Columns a and b use g_1 as their merge sequence (higher value means newer data); columns c and d use g_2 as their merge sequence.

Table widening with aggregation

Within partial-update, you can also apply aggregate functions to specific columns by setting fields.<field-name>.aggregate-function.

Example:

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

Columns a and b use g_1 as their merge sequence (higher value means newer data): a keeps the latest non-null value (defaults to last_non_null_value), b keeps the maximum. Columns c and d use g_2 as their merge sequence: c keeps the latest non-null value, d computes the sum.

Reducing unnecessary changelog

By default, Paimon generates changelog even when the updated value is identical to the previous value. Set 'changelog-producer.row-deduplicate' = 'true' to eliminate such no-op changelog. This applies only to the lookup mechanism.

Use this only when no-op changelog is common, as it introduces additional comparison overhead.