Append Only Table

Append Only Table #

If a table does not have a primary key defined, it is an append-only table by default.

You can only insert a complete record into the table. No delete or update is supported and you cannot define primary keys. This type of table is suitable for use cases that do not require updates (such as log data synchronization).

Bucketing #

You can also define bucket number for Append-only table, see Bucket.

It is recommended that you set the bucket-key field. Otherwise, the data will be hashed according to the whole row, and the performance will be poor.

Compaction #

By default, the sink node will automatically perform compaction to control the number of files. The following options control the strategy of compaction:

Key Default Type Description
write-only
 false Boolean If set to true, compactions and snapshot expiration will be skipped. This option is used along with dedicated compact jobs.
compaction.min.file-num
 5 Integer For file set [f_0,...,f_N], the minimum file number which satisfies sum(size(f_i)) >= targetFileSize to trigger a compaction for append-only table. This value avoids almost-full-file to be compacted, which is not cost-effective.
compaction.max.file-num
 50 Integer For file set [f_0,...,f_N], the maximum file number to trigger a compaction for append-only table, even if sum(size(f_i)) < targetFileSize. This value avoids pending too much small files, which slows down the performance.
full-compaction.delta-commits
 (none) Integer Full compaction will be constantly triggered after delta commits.

Streaming Source #

Streaming source behavior is only supported in Flink engine at present.

Streaming Read Order #

For streaming reads, records are produced in the following order:

  • For any two records from two different partitions
    • If scan.plan-sort-partition is set to true, the record with a smaller partition value will be produced first.
    • Otherwise, the record with an earlier partition creation time will be produced first.
  • For any two records from the same partition and the same bucket, the first written record will be produced first.
  • For any two records from the same partition but two different buckets, different buckets are processed by different tasks, there is no order guarantee between them.

Watermark Definition #

You can define watermark for reading Paimon tables:

CREATE TABLE T (
    `user` BIGINT,
    product STRING,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (...);

-- launch a bounded streaming job to read paimon_table
SELECT window_start, window_end, COUNT(`user`) FROM TABLE(
 TUMBLE(TABLE T, DESCRIPTOR(order_time), INTERVAL '10' MINUTES)) GROUP BY window_start, window_end;

You can also enable Flink Watermark alignment, which will make sure no sources/splits/shards/partitions increase their watermarks too far ahead of the rest:

Key Default Type Description
scan.watermark.alignment.group
 (none) String A group of sources to align watermarks.
scan.watermark.alignment.max-drift
 (none) Duration Maximal drift to align watermarks, before we pause consuming from the source/task/partition.

Bounded Stream #

Streaming Source can also be bounded, you can specify ‘scan.bounded.watermark’ to define the end condition for bounded streaming mode, stream reading will end until a larger watermark snapshot is encountered.

Watermark in snapshot is generated by writer, for example, you can specify a kafka source and declare the definition of watermark. When using this kafka source to write to Paimon table, the snapshots of Paimon table will generate the corresponding watermark, so that you can use the feature of bounded watermark when streaming reads of this Paimon table.

CREATE TABLE kafka_table (
    `user` BIGINT,
    product STRING,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka'...);

-- launch a streaming insert job
INSERT INTO paimon_table SELECT * FROM kakfa_table;

-- launch a bounded streaming job to read paimon_table
SELECT * FROM paimon_table /*+ OPTIONS('scan.bounded.watermark'='...') */;

Example #

The following is an example of creating the Append-Only table and specifying the bucket key.

CREATE TABLE MyTable (
    product_id BIGINT,
    price DOUBLE,
    sales BIGINT
) WITH (
    'bucket' = '8',
    'bucket-key' = 'product_id'
);