diff --git a/docs/docs/append-table/bucketed.mdx b/docs/docs/append-table/bucketed.mdx index 66b78ad57cdc..cf3378630644 100644 --- a/docs/docs/append-table/bucketed.mdx +++ b/docs/docs/append-table/bucketed.mdx @@ -27,16 +27,20 @@ under the License. # Bucketed Append -You can define the `bucket` and `bucket-key` to get a bucketed append table. +A bucketed append table distributes rows using a fixed number of buckets and a bucket key. Rows with the same bucket +key in the same partition are routed to the same bucket. This supports bucket pruning, compatible bucketed joins, and +ordered streaming reads within a bucket. It does not deduplicate rows or create a primary key. -Example to create bucketed append table: +## Create a Bucketed Table - +Use a positive `bucket` count and choose `bucket-key` columns that match your query or ordering requirements. +The following examples assume a configured Paimon catalog and create a separate table named `bucketed_table`. + ```sql -CREATE TABLE my_table ( +CREATE TABLE bucketed_table ( product_id BIGINT, price DOUBLE, sales BIGINT @@ -47,136 +51,86 @@ CREATE TABLE my_table ( ``` + - +```sql +CREATE TABLE bucketed_table ( + product_id BIGINT, + price DOUBLE, + sales BIGINT +) USING paimon +TBLPROPERTIES ( + 'bucket' = '8', + 'bucket-key' = 'product_id' +); +``` -## Data Skipping + + -The primary and most significant advantage of a bucketed append table is **data skipping**. When queries contain -equality (`=`) or `IN` filter conditions on the complete `bucket-key`, Paimon can efficiently push these predicates -down to skip irrelevant bucket files entirely. This means a large number of files that do not match the filter are -pruned before reading, drastically reducing I/O and accelerating queries. +Bucket count controls the distribution, while the bucket key determines which rows are colocated. A skewed key can +concentrate work in a few buckets. More buckets can improve pruning for selective lookups, but can also create more +small files for a small write workload. -For a composite `bucket-key`, the query predicate must cover all bucket-key columns with finite equality or `IN` -values to determine the target buckets. +## Data Skipping -For example, if `bucket-key` is `product_id` and you query: +When a query contains equality (`=`) or `IN` predicates on the complete `bucket-key`, Paimon can compute the candidate +buckets and skip files in the other buckets: ```sql -SELECT * FROM my_table WHERE product_id = 12345; - -SELECT * FROM my_table WHERE product_id IN (1, 2, 3); +SELECT * FROM bucketed_table WHERE product_id = 12345; +SELECT * FROM bucketed_table WHERE product_id IN (1, 2, 3); ``` -Paimon will only read the bucket that contains the matching `product_id` values, filtering out all other bucket files. -This is extremely effective when the table has many buckets and you are querying a small subset of bucket-key values. +An equality lookup reads the matching bucket. An `IN` lookup may read several buckets, and different key values can +map to the same bucket. Rows inside the selected buckets still need to satisfy the query predicate. + +For a composite key such as `bucket-key = product_id,region`, the predicate must constrain **both** columns to finite +equality or `IN` values. Filtering on only `product_id`, or using only a range predicate, does not identify a fixed +set of buckets through this optimization. ## Bucketed Join -Bucketed table can also be used to accelerate join queries by avoiding costly shuffle operations in batch processing. -For example, you can use the following Spark SQL to read a Paimon table: +Spark can use compatible bucket distributions to avoid a shuffle in a batch join. Enable V2 bucketing and join on the +distribution keys: ```sql SET spark.sql.sources.v2.bucketing.enabled = true; -CREATE TABLE FACT_TABLE (order_id INT, f1 STRING) TBLPROPERTIES ('bucket'='10', 'bucket-key' = 'order_id'); +CREATE TABLE fact_table (order_id INT, f1 STRING) USING paimon +TBLPROPERTIES ('bucket' = '10', 'bucket-key' = 'order_id'); -CREATE TABLE DIM_TABLE (order_id INT, f2 STRING) TBLPROPERTIES ('bucket'='10', 'primary-key' = 'order_id'); +CREATE TABLE dim_table (order_id INT, f2 STRING) USING paimon +TBLPROPERTIES ('bucket' = '10', 'primary-key' = 'order_id'); -SELECT * FROM FACT_TABLE AS fact JOIN DIM_TABLE AS dim ON fact.order_id = dim.order_id; +SELECT * FROM fact_table AS fact +JOIN dim_table AS dim ON fact.order_id = dim.order_id; ``` -The `spark.sql.sources.v2.bucketing.enabled` config is used to enable bucketing for V2 data sources. When turned on, -Spark will recognize the specific distribution reported by a V2 data source through SupportsReportPartitioning, and -will try to avoid shuffle if necessary. - -The costly join shuffle will be avoided if two tables have the same bucketing strategy and same number of buckets. +In this example, the fact table is an append table and the dimension table is a primary key table. They use the same +bucket count and compatible distribution keys. Spark uses the partitioning reported by the Paimon source when planning +the join; verify the resulting plan with `EXPLAIN` because the chosen join strategy also depends on Spark's optimizer. ## Bucketed Streaming -An ordinary Append table has no strict ordering guarantees for its streaming writes and reads, but there are some cases -where you need to define a key similar to Kafka's. - -Every record in the same bucket is ordered strictly, streaming read will transfer the record to down-stream exactly in -the order of writing. To use this mode, you do not need to config special configurations, all the data will go into one -bucket as a queue. - -![](/img/for-queue.png) +With `bucket-append-ordered = true` (the default), a streaming reader preserves append order within the **same partition +and bucket**. It provides neither an event-time sort nor a global order across buckets. -**Streaming Read Order** +![Two buckets in one partition preserve their own append order while readers process the buckets independently.](/img/append-bucket-order.svg) -For streaming reads, records are produced in the following order: +To use one queue within each partition, explicitly set `bucket = 1`. Omitting `bucket` selects the unaware-bucket layout, +which has no such ordering guarantee. -* 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. +### Streaming Read Order -**Watermark Definition** +- Within one partition and bucket, earlier appended records are read before later appended records. +- Different buckets can be read by different tasks, so records from those buckets can interleave downstream. +- `scan.plan-sort-partition = true` sorts planned files by partition fields. This is useful when reading the initial + snapshot of a partitioned table; it does not sort individual records by event time or make all downstream tasks + emit in one global order. -You can define watermark for reading Paimon tables: +Keep `bucket-append-ordered = true` when consumers rely on append order. To enable +[incremental clustering](./incremental-clustering) on a bucketed table, set it to `false`; clustering rewrites rows in +clustering order and gives up that append-order guarantee. -```sql -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](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/event-time/generating_watermarks/#watermark-alignment-_beta_), -which will make sure no sources/splits/shards/partitions increase their watermarks too far ahead of the rest: - - - - - - - - - - - - - - - - - - - - - - - - -
KeyDefaultTypeDescription
scan.watermark.alignment.group
(none)StringA group of sources to align watermarks.
scan.watermark.alignment.max-drift
(none)DurationMaximal 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. - -```sql -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'='...') */; -``` +For scan startup modes, watermarks, alignment, and bounded streaming reads, see [Streaming](./streaming). diff --git a/docs/docs/append-table/incremental-clustering.mdx b/docs/docs/append-table/incremental-clustering.mdx index 688ee58290c8..25249a8f167c 100644 --- a/docs/docs/append-table/incremental-clustering.mdx +++ b/docs/docs/append-table/incremental-clustering.mdx @@ -1,6 +1,6 @@ --- title: "Incremental Clustering" -sidebar_position: 2 +sidebar_position: 4 --- import Tabs from '@theme/Tabs'; @@ -27,268 +27,275 @@ under the License. # Incremental Clustering -Paimon currently supports ordering append tables using SFC (Space-Filling Curve)(see [sort compact](../maintenance/dedicated-compaction#sort-compact) for more info). -The resulting data layout typically delivers better performance for queries that target clustering keys. -However, with the current SortCompaction, even when neither the data nor the clustering keys have changed, -each run still rewrites the entire dataset, which is extremely costly. +Incremental clustering improves the data layout of append tables by sorting selected files on frequently filtered +columns. Compared with repeatedly sorting an entire partition, it can reduce the amount of data rewritten while +improving [file-statistics pruning](./query-performance#file-statistics-and-clustering). A run may select no files when +its compaction criteria are not met. Full mode considers all runs in the selected scope, but can skip work that is +already clustered; see [file selection](#implement). -To address this, Paimon introduced a more flexible, incremental clustering mechanism—Incremental Clustering. -On each run, it selects only a specific subset of files to cluster, avoiding a full rewrite. This enables low-cost, -sort-based optimization of the data layout and improves query performance. In addition, with Incremental Clustering, -you can adjust clustering keys without rewriting existing data, the layout evolves dynamically as cluster runs and -gradually converges to an optimal state, significantly reducing the decision-making complexity around data layout. +Clustering also merges small files, respecting `target-file-size`. It changes the physical layout, not the rows returned +by a query, and does not replace SQL `ORDER BY`. +## Requirements -Incremental Clustering supports: -- Support incremental clustering; minimizing write amplification as possible. -- Support small-file compaction; during rewrites, respect target-file-size. -- Support changing clustering keys; newly ingested data is clustered according to the latest clustering keys. -- Provide a full mode; when selected, the entire dataset will be reclustered. +| Requirement | Unaware-bucket append (`bucket = -1`) | Bucketed append (`bucket > 0`) | +| --- | --- | --- | +| Primary key | Must not be defined. | Must not be defined. | +| Enable clustering | `clustering.incremental = true` and nonempty `clustering.columns`. | Same. | +| Append ordering | No bucket-order guarantee. | Must set `bucket-append-ordered = false`. | +| Deletion vectors | Supported. | Must remain disabled. | +| Compaction execution | Schedule explicit clustering jobs; the Flink sink's normal background compaction is disabled. | Writer compaction and dedicated compact jobs use the bucket clustering path. | +| Global/local sort mode | Configurable for batch clustering jobs. | Clustering is performed within each partition and bucket; the global/local option does not select this path. | +| Historical-partition auto-clustering | Supported. | `clustering.history-partition.*` does not apply. | -Incremental Clustering is supported for append tables in both unaware-bucket mode (`bucket = -1`) and -bucketed mode (`bucket > 0`). For bucketed append tables, additional requirements apply because -clustering gives up the ordered append guarantee within buckets. +Data Evolution tables cannot enable incremental clustering. If streaming consumers require ordered append reads from a +bucketed table, keep that ordering and do not enable clustering. ## Enable Incremental Clustering -To enable Incremental Clustering, the following configuration needs to be set for the table: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
OptionValueRequiredTypeDescription
clustering.incremental
trueYesBooleanMust be set to true to enable incremental clustering. Default is false.
clustering.columns
'clustering-columns'YesStringThe clustering columns, in the format 'columnName1,columnName2'. It is not recommended to use partition keys as clustering keys.
clustering.strategy
'zorder' or 'hilbert' or 'order'NoStringThe ordering algorithm used for clustering. If not set, It'll decided from the number of clustering columns. 'order' is used for 1 column, 'zorder' for less than 5 columns, and 'hilbert' for 5 or more columns.
clustering.incremental.mode
'global-sort' or 'local-sort'NoEnumThe sort mode for incremental clustering compaction. Default is global-sort. global-sort performs a global range shuffle across tasks before local sorting, output files are globally ordered by the clustering columns at the cost of network shuffling. local-sort skips the global shuffle and sorts rows only within each compaction task independently, each output file is internally ordered but there is no global ordering across files, this mode is cheaper and sufficient for per-file Parquet lookup optimizations.
- -For bucketed append tables (`bucket > 0`), you must also set the following option: - - - - - - - - - - - - - - - - - - - - - -
OptionValueRequiredTypeDescription
bucket-append-ordered
falseYesBooleanMust be set to false for bucketed append tables with incremental clustering.
- -Bucketed append tables with Incremental Clustering do not support `deletion-vectors.enabled = true`. - -Example: +Set the clustering keys on the table using the DDL for its layout. Choose the tab for your engine. + +### Unaware-Bucket Table + +For `my_table` from the [overview](./), enable clustering and specify the columns: + + + + +```sql +ALTER TABLE my_table SET ( + 'clustering.incremental' = 'true', + 'clustering.columns' = 'product_id,price' +); +``` + + + ```sql -ALTER TABLE T SET ( +ALTER TABLE my_table SET TBLPROPERTIES ( + 'clustering.incremental' = 'true', + 'clustering.columns' = 'product_id,price' +); +``` + + + + +### Bucketed Table + +For `bucketed_table` from [Bucketed append](./bucketed#create-a-bucketed-table), disable append ordering in the **same** +statement that enables clustering. Keep deletion vectors disabled. This explicitly opts out of +[ordered streaming reads](./bucketed#bucketed-streaming). + + + + +```sql +ALTER TABLE bucketed_table SET ( 'bucket-append-ordered' = 'false', 'clustering.incremental' = 'true', - 'clustering.columns' = 'event_time,user_id', - 'clustering.strategy' = 'zorder' + 'clustering.columns' = 'product_id,price' ); ``` -Once Incremental Clustering for a table is enabled, you can run Incremental Clustering in batch mode periodically -to continuously optimizes data layout of the table and deliver better query performance. + + -**Note**: Since common compaction also rewrites files, it may disrupt the ordered data layout built by Incremental Clustering. -Therefore, when Incremental Clustering is enabled, the table no longer supports write-time compaction or dedicated compaction; -clustering and small-file merging must be performed exclusively via Incremental Clustering runs. +```sql +ALTER TABLE bucketed_table SET TBLPROPERTIES ( + 'bucket-append-ordered' = 'false', + 'clustering.incremental' = 'true', + 'clustering.columns' = 'product_id,price' +); +``` -## Run Incremental Clustering -:::info + + -The following examples submit batch compact jobs. They are the recommended way to run Incremental Clustering explicitly. +### Clustering Options -::: +| Option | Default | How to use it | +| --- | --- | --- | +| `clustering.incremental` | `false` | Set to `true` to enable incremental clustering. | +| `clustering.columns` | Not set | Comma-separated columns, such as `product_id,price`. Prefer frequently filtered data columns over partition columns. | +| `clustering.strategy` | `auto` | `order`, `zorder`, or `hilbert`. Automatic selection uses `order` for one column, `zorder` for two to four, and `hilbert` for five or more. | +| `clustering.incremental.mode` | `global-sort` | Sort execution mode for unaware-bucket batch clustering; see below. | -To run a Incremental Clustering job, follow these instructions. +## Choose a Sort Mode -You don't need to specify any clustering-related parameters when running Incremental Clustering, -these options are already defined as table options. If you need to change clustering settings, please update the corresponding table options. +For unaware-bucket tables, the mode controls how the **selected files in each partition** are sorted: - +| Mode | Execution | Tradeoff | +| --- | --- | --- | +| `global-sort` | Range-shuffles rows across tasks, then sorts within tasks using the configured clustering strategy. | Coordinates the layout across the selected output files, at the cost of a network shuffle. | +| `local-sort` | Sorts rows independently within each compaction task, without the global range shuffle. | Less shuffle work; ranges in files produced by different tasks can overlap. Useful when ordering within files is sufficient, such as for Parquet lookup optimizations. | - +Here, “global” refers to the selected clustering work within a partition. It does not imply that all existing files or +all table partitions become globally ordered after an incremental run. -Run the following sql: +## Run Incremental Clustering + +Run explicit compact jobs in batch mode. Table options supply the clustering columns and strategy. The examples below +show routine incremental selection (`minor`) and full clustering (`full`) of the selected table or partition scope. + + + ```sql ---set the write parallelism, if too big, may generate a large number of small files. -SET spark.sql.shuffle.partitions=10; +-- Choose parallelism for the workload; too many tasks can produce small files. +SET spark.sql.shuffle.partitions = 10; --- run incremental clustering -CALL sys.compact(table => 'T') +-- Select files using the incremental compaction strategy. +CALL sys.compact(table => 'my_table', compact_strategy => 'minor'); --- run incremental clustering with full mode, this will recluster all data -CALL sys.compact(table => 'T', compact_strategy => 'full') +-- Alternatively, request full clustering; already-clustered runs can be skipped. +CALL sys.compact(table => 'my_table', compact_strategy => 'full'); +``` --- run incremental clustering with global-sort mode (default) --- performs a global range shuffle across tasks, output files are globally ordered -CALL sys.compact(table => 'T', options => 'clustering.incremental.mode=global-sort') +With historical-partition auto-clustering disabled (the default), use `partitions` to limit the work to one partition +of `my_table`: --- run incremental clustering with local-sort mode --- sorts rows only within each task, no global shuffle, cheaper and sufficient for Parquet lookup optimizations -CALL sys.compact(table => 'T', options => 'clustering.incremental.mode=local-sort') +```sql +CALL sys.compact( + table => 'my_table', + partitions => 'dt=2026-09-10', + compact_strategy => 'full' +); ``` - +Alternatively, `where` accepts a predicate on partition columns. Do not combine `partitions` and `where` in one call. - +On unaware-bucket tables, [historical-partition auto-clustering](#auto-clustering-for-historical-partition) can add full +clustering of partitions outside either filter, so these arguments are not a hard job boundary when that feature is +enabled. -Run the following command to submit a incremental clustering job for the table. +For the unpartitioned `bucketed_table` example, change the table name and omit the partition filter. + +For an unaware-bucket table, you can override the sort mode for one invocation: + +```sql +CALL sys.compact( + table => 'my_table', + compact_strategy => 'minor', + options => 'clustering.incremental.mode=local-sort' +); +``` + + + ```bash /bin/flink run \ + -Dexecution.runtime-mode=batch \ /path/to/paimon-flink-action-@@VERSION@@.jar \ compact \ --warehouse \ --database \ --table \ - [--compact_strategy ] \ - [--table_conf ] \ - [--catalog_conf [--catalog_conf ...]] + --compact_strategy minor \ + --table_conf sink.parallelism=2 ``` -Example: run incremental clustering with global-sort mode (default), output files are globally ordered across all tasks. +Replace the placeholders with your installation and table details. Use `--compact_strategy full` for full clustering. +For the partitioned `my_table` example, add `--partition dt=2026-09-10` to select that partition. This limits the whole +job when [historical-partition auto-clustering](#auto-clustering-for-historical-partition) is disabled; when enabled, +the job can also fully cluster eligible historical partitions outside the requested partition. +For an unaware-bucket table, add `--table_conf clustering.incremental.mode=local-sort` to override the default sort mode. +Add `--catalog_conf key=value` arguments as required by your catalog and storage. -```bash -/bin/flink run \ - /path/to/paimon-flink-action-@@VERSION@@.jar \ - compact \ - --warehouse s3:///path/to/warehouse \ - --database test_db \ - --table test_table \ - --table_conf sink.parallelism=2 \ - --table_conf clustering.incremental.mode=global-sort \ - --compact_strategy minor \ - --catalog_conf s3.endpoint=https://****.com \ - --catalog_conf s3.access-key=***** \ - --catalog_conf s3.secret-key=***** +`sink.parallelism` controls the Flink action's sink parallelism. Size it for the amount of selected data rather than +using a large value for every run. + + + + +The `compact` entry points route to the appropriate clustering implementation when clustering is enabled. On +unaware-bucket tables, use these jobs for recurring small-file merging and layout maintenance. On bucketed tables, +writer compaction can also perform incremental clustering; set `write-only = true` on ingestion if that work should +be performed only by dedicated jobs. + +## Verify the Result + +Inspect the files before and after a clustering job. For example, in Spark SQL: + +```sql +SELECT `partition`, bucket, level, + COUNT(*) AS file_count, + SUM(file_size_in_bytes) AS total_bytes +FROM `my_table$files` +GROUP BY `partition`, bucket, level +ORDER BY `partition`, bucket, level; + +SELECT snapshot_id, commit_kind, commit_time +FROM `my_table$snapshots` +ORDER BY snapshot_id DESC +LIMIT 10; ``` -Example: run incremental clustering with local-sort mode, sorts rows only within each task without global shuffle, cheaper and sufficient for Parquet lookup optimizations. +The [files system table](../concepts/system-tables#files-table) shows the current physical layout. Compare file counts, +sizes, and levels in the partitions and buckets selected by the job. The snapshots table helps identify commits around +the job's execution time; concurrent ingestion can also create snapshots. A successful job may leave the files unchanged +when the planner selects no work, including the [full-mode skip cases](#implement). -```bash -/bin/flink run \ - /path/to/paimon-flink-action-@@VERSION@@.jar \ - compact \ - --warehouse s3:///path/to/warehouse \ - --database test_db \ - --table test_table \ - --table_conf sink.parallelism=2 \ - --table_conf clustering.incremental.mode=local-sort \ - --compact_strategy minor \ - --catalog_conf s3.endpoint=https://****.com \ - --catalog_conf s3.access-key=***** \ - --catalog_conf s3.secret-key=***** +To inspect pruning potential, query `file_path`, `min_value_stats`, and `max_value_stats` from `my_table$files` and compare +the ranges for your filter columns. Then run a representative query and compare scan metrics such as files or bytes +read. Higher levels or fewer files alone do not establish that a query reads less data. Keep the data and query +comparable when measuring the effect, especially if ingestion continues during maintenance. + +## Change Clustering Keys + +Update `clustering.columns` and, if needed, `clustering.strategy` using the same `ALTER TABLE` syntax as above. Changing +the options does not immediately rewrite existing data. Subsequent clustering uses the new settings for selected files. +After changing the clustering columns, use `compact_strategy = full` to apply them across the selected scope. +Changing only `clustering.strategy` does not force an already-clustered run to be rewritten: the full-mode skip check +compares the clustering columns, not the sorting strategy. + +## Auto-Clustering for Historical Partitions {#auto-clustering-for-historical-partition} + +For partitioned unaware-bucket tables, a clustering run can also select inactive historical partitions **outside the +requested partition predicate** for full clustering. This additional selection requires both a configured idle duration +and an explicit partition predicate: Spark's `partitions` or `where`, or Flink's `--partition`. + +Without an explicit partition predicate, the historical auto-full path is inactive. For example, setting the idle +duration does not activate it for the unscoped `minor` call above; that call still uses normal incremental selection +across the table. Auto-clustering happens within a submitted job; configuring the options does not start a scheduler. + +| Option | Default | Purpose | +| --- | --- | --- | +| `clustering.history-partition.idle-to-full-sort` | Not set (disabled) | How long a partition must have no new updates before it is considered historical. | +| `clustering.history-partition.limit` | `5` | Maximum number of additional historical partitions selected outside the requested partition predicate in a run. | + +For example, in Flink SQL: + +```sql +ALTER TABLE my_table SET ( + 'clustering.history-partition.idle-to-full-sort' = '3 d', + 'clustering.history-partition.limit' = '5' +); ``` -* `--compact_strategy` Determines how to pick files to be cluster, the default is `minor`. - * `full` : All files will be selected for clustered. - * `minor` : Pick the set of files that need to be clustered based on specified conditions. -Note: write parallelism is set by `sink.parallelism`, if too big, may generate a large number of small files. +With these options, a job requesting `dt=2026-09-10` can also fully cluster up to five eligible historical partitions +outside that date. The additional partitions use full clustering even if the requested partitions use +`compact_strategy = minor`. The planner evaluates their eligibility and whether a full rewrite is needed. -You can use `-D execution.runtime-mode=batch` or `-yD execution.runtime-mode=batch` (for the ON-YARN scenario) to use batch mode. +To keep the entire job within the requested partitions, leave `clustering.history-partition.idle-to-full-sort` unset, +or remove it from the table options before submitting the job. The partition limit controls the additional historical +work, not the number of explicitly requested partitions. These options do not apply to bucketed append tables. - +## How File Selection Works {#implement} - +Incremental clustering organizes files into levels and uses a universal compaction strategy to select sorted runs. +Normally, newly appended files enter level 0. At level 0, each file is treated as its own run; at a higher level, the +files produced by a clustering set are treated as one run. + +![Incremental clustering selects new files and an eligible existing run, rewrites that set, and keeps an unselected higher-level run.](/img/append-incremental-clustering.svg) + +The planner balances the number and relative sizes of runs against write amplification. A run rewrites only the +selected set, so higher-level files can remain untouched. The levels in the diagram are illustrative; the selected +files and output level depend on the planner. -## Auto-Clustering For Historical Partition -:::info - -Auto-clustering for historical partitions currently applies only to unaware-bucket append tables (`bucket = -1`). -Bucketed append clustering does not use the `clustering.history-partition.*` table options. - -::: - -While performing incremental clustering on recently active partitions, Paimon can automatically detect historical and -inactive partitions and evaluate whether their data layout has reached an optimal state. -For those historical partitions that have not yet achieved optimal layout, Paimon will also perform full clustering on them -during the same operation, thereby improving their query performance. - -To enable auto-clustering for historical partitions, the following configuration needs to be set for the table: - - - - - - - - - - - - - - - - - - - - - - - - - - - -
OptionValueRequiredTypeDescription
clustering.history-partition.idle-to-full-sort
3dYesDurationThe duration after which a partition without new updates is considered a historical partition. Default is null.
clustering.history-partition.limit
5YesIntegerThe limit of history partition number for automatically performing full clustering. Default value is 5.
- - -## Implement -To balance write amplification and sorting effectiveness, Paimon leverages the LSM Tree notion of levels to stratify data files -and uses the Universal Compaction strategy to select files for clustering. -- Newly written data lands in level-0; files in level-0 are unclustered. -- All files in level-i are produced by sorting within the same sorting set. -- By analogy with Universal Compaction: in level-0, each file is a sorted run; in level-i, all files together constitute a single sorted run. During clustering, the sorted run is the basic unit of work. - -By introducing more levels, we can control the amount of data processed in each clustering run. -Data at higher levels is more stably clustered and less likely to be rewritten, thereby mitigating write amplification while maintaining good sorting effectiveness. +Full mode selects all runs in a compaction unit unless there is no data, or the unit already contains a single run at +the highest level with the same clustering columns. In those cases, no rewrite is needed by the planner. This check +applies per partition for unaware-bucket tables and per partition and bucket for bucketed tables, so a full job can +rewrite some units while leaving others unchanged. diff --git a/docs/docs/append-table/index.mdx b/docs/docs/append-table/index.mdx index 152cb9c648a0..9671911e8f92 100644 --- a/docs/docs/append-table/index.mdx +++ b/docs/docs/append-table/index.mdx @@ -27,156 +27,102 @@ under the License. # Overview -If a table does not have a primary key defined, it is an append table. Compared to the primary key table, it does not -have the ability to directly receive changelogs. It cannot be directly updated with data through upsert. It can only -receive incoming data from append data. +An append table has no primary key. Each inserted row is stored as a new record, including rows whose values duplicate +an existing record. Inserts do not perform key-based deduplication or upserts. Use a +[primary key table](../primary-key-table/) when incoming changelog records should update existing rows by key. - +Append tables support batch and streaming workloads, with snapshots, time travel, schema evolution, and file-level +query optimizations. You can also change stored rows with explicit [row-level operations](./row-level-operations). +## Create an Append Table + +The examples below assume that you have configured and selected a Paimon catalog. See the +[Flink quick start](../flink/quick-start) or [Spark quick start](../spark/quick-start) for catalog setup. + + ```sql CREATE TABLE my_table ( product_id BIGINT, price DOUBLE, - sales BIGINT -) WITH ( - -- 'target-file-size' = '256 MB', - -- 'file.format' = 'parquet', - -- 'file.compression' = 'zstd', - -- 'file.compression.zstd-level' = '3' + sales BIGINT, + dt STRING +) PARTITIONED BY (dt) WITH ( + 'bucket' = '-1' ); -``` - - - - - -Batch write and batch read in typical application scenarios, similar to a regular Hive partition table, but compared to -the Hive table, it can bring: - -1. Time travel enables reproducible queries that use exactly the same table snapshot, or lets users easily examine - changes. Version rollback allows users to quickly correct problems by resetting tables to a good state. -2. Scan planning is fast — data files are pruned with partition and column-level stats, using table metadata. File - Index (BloomFilter, Bitmap, Range Bitmap) and aggregate push-down further accelerate queries. -3. Schema evolution supports add, drop, update, or rename columns, and has no side-effects. -4. Rich ecosystem — adds tables to compute engines including Flink, Spark, Hive, Trino, Presto, StarRocks, and Doris, - working just like a SQL table. -5. Incremental Clustering with z-order/hilbert/order sorting to optimize data layout at low cost. -6. Streaming read & write like a queue, DELETE / UPDATE / MERGE INTO support low-cost row-level operations. - -## Append Streaming - -You can stream write to the Append table in a very flexible way through Flink, or read the Append table through -Flink, using it like a queue. The only difference is that its latency is in minutes. Its advantages are very low cost -and the ability to push down filters and projection. - -**Pre small files merging** - -"Pre" means that this compact occurs before committing files to the snapshot. - -If Flink's checkpoint interval is short (for example, 30 seconds), each snapshot may produce lots of small changelog -files. Too many files may put a burden on the distributed storage cluster. - -In order to compact small changelog files into large ones, you can set the table option `precommit-compact = true`. -Default value of this option is false, if true, it will add a compact coordinator and worker operator after the -writer operator, which copies changelog files into large ones. - -**Post small files merging** - -"Post" means that this compact occurs after committing files to the snapshot. - -In streaming write job, without bucket definition, there is no compaction in writer, instead, will use -`Compact Coordinator` to scan the small files and pass compaction task to `Compact Worker`. In streaming mode, if you -run insert sql in flink, the topology will be like this: - -![](/img/unaware-bucket-topo.png) - -Do not worry about backpressure, compaction never backpressure. -If you set `write-only` to true, the `Compact Coordinator` and `Compact Worker` will be removed in the topology. - -The auto compaction is only supported in Flink engine streaming mode. You can also start a compaction job in Flink by -Flink action in Paimon and disable all the other compactions by setting `write-only`. - -**Streaming Query** - -You can stream the Append table and use it like a Message Queue. As with primary key tables, there are two options -for streaming reads: -1. By default, Streaming read produces the latest snapshot on the table upon first startup, and continue to read the - latest incremental records. -2. You can specify `scan.mode`, `scan.snapshot-id`, `scan.timestamp-millis` and/or `scan.file-creation-time-millis` to - stream read incremental only. - -Similar to flink-kafka, order is not guaranteed by default, if your data has some sort of order requirement, you also -need to consider defining a `bucket-key`, see [Bucketed Append](./bucketed) - -## Aggregate push down - -Append Table supports aggregate push down: - -```sql -SELECT COUNT(*) FROM TABLE WHERE DT = '20230101'; +INSERT INTO my_table VALUES (1, 10.0, 2, '2026-09-10'); +SELECT * FROM my_table WHERE dt = '2026-09-10'; ``` -This query can be accelerated during compilation and returns very quickly. - -For Spark SQL, table with default `metadata.stats-mode` can be accelerated: + + ```sql -SELECT MIN(a), MAX(b) FROM TABLE WHERE DT = '20230101'; +CREATE TABLE my_table ( + product_id BIGINT, + price DOUBLE, + sales BIGINT, + dt STRING +) USING paimon +PARTITIONED BY (dt) +TBLPROPERTIES ( + 'bucket' = '-1' +); -SELECT * FROM TABLE ORDER BY a LIMIT 1; +INSERT INTO my_table VALUES (1, 10.0, 2, '2026-09-10'); +SELECT * FROM my_table WHERE dt = '2026-09-10'; ``` -Min max topN query can be also accelerated during compilation and returns very quickly. + + -## Data Skipping By Order +`bucket = -1` is the default for append tables; it is shown explicitly here to identify the layout. Partitioning is +optional. File size, format, and compression can be configured with `target-file-size`, `file.format`, and +`file.compression`; see [Configurations](../maintenance/configurations). -Paimon by default records the maximum and minimum values of each field in the manifest file. +## Choose a Layout -In the query, according to the `WHERE` condition of the query, together with the statistics in the manifest we can -perform file filtering. If the filtering effect is good, the query that would have cost minutes will be accelerated to -milliseconds to complete the execution. +| Layout | Configuration | Use it when | Considerations | +| --- | --- | --- | --- | +| Unaware-bucket append | `bucket = -1` (default) | You want flexible ingestion without distributing rows by a bucket key. | No streaming row-order guarantee. Supports row tracking. | +| [Bucketed append](./bucketed) | Positive `bucket` and a `bucket-key` | Queries filter on bucket keys, joins can reuse the distribution, or streaming consumers need ordering within a bucket. | Ordering is scoped to one partition and bucket, and requires `bucket-append-ordered = true`. | -Often the data distribution is not always ideal for filtering, so can we sort the data by the field in `WHERE` condition? -You can take a look at [Flink COMPACT Action](../maintenance/dedicated-compaction#sort-compact), -[Flink COMPACT Procedure](../flink/procedures) or [Spark COMPACT Procedure](../spark/procedures). +[Incremental clustering](./incremental-clustering) is a data-layout optimization available for both layouts. On bucketed +append tables it requires giving up the ordered append guarantee. Bucketing, clustering, and row tracking solve different +problems; choose them according to your query and ingestion requirements. -## Data Skipping By File Index +## Read and Maintain the Table -You can use file index too, it filters files by indexing on the reading side. +Examples named `my_table` use the schema above. Individual guides introduce separate tables when they need a different +layout or schema. -Define `file-index.bitmap.columns`, Data file index is an external index file and Paimon will create its -corresponding index file for each file. If the index file is too small, it will be stored directly in the manifest, -otherwise in the directory of the data file. Each data file corresponds to an index file, which has a separate file -definition and can contain different types of indexes with multiple columns. +### Streaming {#append-streaming} -Different file indexes may be efficient in different scenarios. For example bloom filter may speed up query in point lookup -scenario. Using a bitmap may consume more space but can result in greater accuracy. +[Streaming](./streaming) covers Flink ingestion, small-file compaction, scan startup modes, and watermarks. +For ordering requirements, see [Bucketed streaming](./bucketed#bucketed-streaming). -* [BloomFilter](../concepts/spec/fileindex#index-bloomfilter): `file-index.bloom-filter.columns`. -* [Bitmap](../concepts/spec/fileindex#index-bitmap): `file-index.bitmap.columns`. -* [Range Bitmap](../concepts/spec/fileindex#index-range-bitmap): `file-index.range-bitmap.columns`. +### Query Performance -If you want to add file index to existing table, without any rewrite, you can use `rewrite_file_index` procedure. Before -we use the procedure, you should config appropriate configurations in target table. You can use ALTER clause to config -`file-index..columns` to the table. +#### Aggregate Pushdown {#aggregate-push-down} -How to invoke: see [flink procedures](../flink/procedures) +Some aggregate queries can use metadata instead of reading every data row. See +[Aggregate pushdown](./query-performance#aggregate-pushdown) for examples and limits. -## Row Level Operations +#### Clustering and File Statistics {#data-skipping-by-order} -Now, only Spark SQL supports DELETE & UPDATE & MERGE INTO, you can take a look at [Spark Write](../spark/sql-write). +Sorting can narrow the value ranges in each file and improve pruning. Start with +[File statistics and clustering](./query-performance#file-statistics-and-clustering), then configure +[Incremental clustering](./incremental-clustering) for recurring optimization. -Example: -```sql -DELETE FROM my_table WHERE currency = 'UNKNOWN'; -``` +#### File Indexes {#data-skipping-by-file-index} + +[File indexes](./query-performance#file-indexes) provide additional filtering for supported predicates, including +Bloom filter, bitmap, and range bitmap indexes. -Update append table has two modes: +### Row-Level Operations {#row-level-operations} -1. COW (Copy on Write): search for the hit files and then rewrite each file to remove the data that needs to be deleted - from the files. This operation is costly. -2. MOW (Merge on Write): By specifying `'deletion-vectors.enabled' = 'true'`, the Deletion Vectors mode can be enabled. - Only marks certain records of the corresponding file for deletion and writes the deletion file, without rewriting the entire file. +[Row-level operations](./row-level-operations) explains Spark SQL `DELETE`, `UPDATE`, and `MERGE INTO`, including the +choice between rewriting data files and using deletion vectors. [Row tracking](./row-tracking) adds hidden row IDs +and row versions for tracking changes across these operations. diff --git a/docs/docs/append-table/query-performance.md b/docs/docs/append-table/query-performance.md new file mode 100644 index 000000000000..86e9ebe1f7e6 --- /dev/null +++ b/docs/docs/append-table/query-performance.md @@ -0,0 +1,145 @@ +--- +title: "Query Performance" +sidebar_position: 2 +--- + + + +# Query Performance + +Append-table queries can avoid work at several layers. Start with the predicates your queries use, then choose a +layout or index that helps eliminate irrelevant data. + +| Query pattern | Optimization | What it can skip | +| --- | --- | --- | +| Filters on partition columns | Partition pruning | Other partitions. | +| Equality or `IN` on all bucket-key columns | [Bucket pruning](./bucketed#data-skipping) | Other buckets in a bucketed table. | +| Selective filters on data columns | File statistics, improved by clustering | Files whose value ranges cannot match. | +| Predicates supported by a file index | Bloom filter, bitmap, or range bitmap indexes | Irrelevant data identified by the index. | +| Supported aggregates with sufficient metadata | Aggregate pushdown | Reading data rows to compute the aggregate. | + +These optimizations can be combined. Their effectiveness depends on the engine, the predicate, and the distribution of +the stored values. Use the engine's `EXPLAIN` output and scan metrics to check which optimizations a query actually uses. + +## File Statistics and Clustering + +Paimon stores column statistics in file metadata, subject to the table's statistics configuration. Min/max values let +a reader reject files whose ranges do not overlap a query predicate. For example: + +```sql +SELECT * FROM my_table +WHERE dt = '2026-09-10' AND product_id BETWEEN 100 AND 200; +``` + +The partition filter first limits the scan to one date. Within that partition, file statistics can exclude files +whose `product_id` range falls outside the requested interval. If every file contains a broad mix of product IDs, +min/max pruning will be less effective. + +![The same nine product IDs are spread across three files before clustering. After clustering, only one file overlaps product IDs 100 through 200.](/img/append-file-pruning.svg) + +In this example, every unsorted file overlaps the predicate, so all three must be read. After clustering, only the +middle file overlaps it. File pruning selects candidate files; the reader still evaluates the predicate on their rows. + +[Incremental clustering](./incremental-clustering) sorts selected files by frequently filtered columns, which can make +their value ranges more selective without rewriting every file on each run. Use +[sort compaction](../maintenance/dedicated-compaction#sort-compact) for an explicit sort rewrite. Clustering improves the +physical layout; SQL result ordering still requires `ORDER BY`. + +## File Indexes + +File indexes provide filtering beyond min/max statistics. Configure the relevant columns before writing indexed data: + +| Index | Table option | Typical use | +| --- | --- | --- | +| [Bloom filter](../concepts/spec/fileindex#index-bloomfilter) | `file-index.bloom-filter.columns` | Equality lookups; false positives may still require reading data. | +| [Bitmap](../concepts/spec/fileindex#index-bitmap) | `file-index.bitmap.columns` | Equality and set-membership filtering. | +| [Range bitmap](../concepts/spec/fileindex#index-range-bitmap) | `file-index.range-bitmap.columns` | Range filtering. | + +Each indexed data file has associated index data. Small indexes can be embedded in the manifest; larger indexes are +stored alongside the data files. Indexes add storage and write work, so select columns used by relevant queries. + +### Index Existing Files + +Changing the table options affects subsequently written files; it does not add indexes to existing files. After setting +the options, run `rewrite_file_index` to build indexes for existing data without rewriting the data files. The procedure +still reads the data needed to construct the indexes. + +For example, in Flink SQL, with the table in the `default` database: + +```sql +ALTER TABLE my_table SET ('file-index.bloom-filter.columns' = 'product_id'); +CALL sys.rewrite_file_index(`table` => 'default.my_table'); +``` + +The equivalent Spark SQL is: + +```sql +ALTER TABLE my_table SET TBLPROPERTIES ('file-index.bloom-filter.columns' = 'product_id'); +CALL sys.rewrite_file_index(table => 'default.my_table'); +``` + +Use the actual database name for your table. To limit the rewrite, Flink accepts a `partitions` argument and Spark +accepts a partition predicate in `where`. See [Flink procedures](../flink/procedures) and +[Spark procedures](../spark/procedures) for those engine-specific arguments. + +### Check Index Coverage + +Query the [file indexes system table](../concepts/system-tables#file-indexes-table) to see which data files have the +configured indexes. For example, in Spark SQL: + +```sql +SELECT column_name, index_type, storage_type, + COUNT(DISTINCT file_path) AS indexed_file_count +FROM `my_table$file_indexes` +GROUP BY column_name, index_type, storage_type +ORDER BY column_name, index_type, storage_type; +``` + +Each row in the system table describes one column and index type in one data file. `storage_type` is `EMBEDDED` for +index data stored in metadata and `FILE` for an external index file. Compare the indexed files with `my_table$files` +to check coverage; the presence of an index alone does not show that a query used it. Check the query's predicates, +plan, and scan metrics as well. + +## Aggregate Pushdown + +Supported aggregate queries can use table metadata. Using the partitioned table from the [overview](./): + +```sql +SELECT COUNT(*) FROM my_table WHERE dt = '2026-09-10'; +``` + +Spark can also use column statistics for supported `MIN` and `MAX` queries: + +```sql +SELECT MIN(price), MAX(sales) FROM my_table WHERE dt = '2026-09-10'; +``` + +Keep the required statistics available through `metadata.stats-mode` and any column-specific statistics settings. +Pushdown depends on the query and available metadata: a filter that needs row-by-row evaluation can prevent a +metadata-only aggregate. These examples use a partition predicate so that complete files can be selected. + +Spark can also use statistics to reduce the files scanned for a top-N query: + +```sql +SELECT * FROM my_table ORDER BY price LIMIT 1; +``` + +Top-N pruning does not imply that the full result can be produced from metadata alone. Check the query plan rather +than assuming every aggregate or `ORDER BY ... LIMIT` query avoids reading data files. diff --git a/docs/docs/append-table/row-level-operations.md b/docs/docs/append-table/row-level-operations.md new file mode 100644 index 000000000000..26945b0ff934 --- /dev/null +++ b/docs/docs/append-table/row-level-operations.md @@ -0,0 +1,112 @@ +--- +title: "Row-Level Operations" +sidebar_position: 5 +--- + + + +# Row-Level Operations + +An append table stores each inserted row without key-based deduplication, but you can still modify stored data with +explicit Spark SQL `DELETE`, `UPDATE`, and `MERGE INTO` statements. These operations locate matching rows; they do not +turn ordinary inserts into upserts. + +The examples below use a Paimon catalog and the `my_table` schema from the [overview](./). Configure Spark with +`org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions` as shown in the [Spark quick start](../spark/quick-start). +See [Spark SQL write](../spark/sql-write) for statement syntax. + +## Delete, Update, and Merge + +```sql +DELETE FROM my_table WHERE price < 0; + +UPDATE my_table SET price = 12.0 +WHERE product_id = 1 AND dt = '2026-09-10'; +``` + +For a merge, create a source with the same columns and match on the columns that identify rows in your application: + +```sql +CREATE TEMPORARY VIEW updates AS +SELECT CAST(1 AS BIGINT) AS product_id, CAST(12.0 AS DOUBLE) AS price, + CAST(3 AS BIGINT) AS sales, '2026-09-10' AS dt; + +MERGE INTO my_table AS target +USING updates AS source +ON target.product_id = source.product_id AND target.dt = source.dt +WHEN MATCHED THEN UPDATE SET + target.price = source.price, target.sales = source.sales +WHEN NOT MATCHED THEN INSERT *; +``` + +There is no primary key constraint on this table. In the example, matching uses `(product_id, dt)`: + +- If several target rows have the same matching values, one source row can update all of them. +- With a `WHEN MATCHED` clause, a target row must not match more than one source row. Paimon rejects that ambiguous + match, so resolve duplicate source keys before the merge. +- Source rows that do not match the target are inserted independently. The merge does not deduplicate those rows. + +Choose the merge condition and prepare source rows to express the intended matching behavior. A bucket key does not +enforce uniqueness either. + +## Choose How Changes Are Stored + +For regular append tables, two approaches are available: + +| Approach | Configuration | What happens | +| --- | --- | --- | +| Copy on write (COW) | Deletion vectors disabled (default). | Affected files are replaced with files containing the surviving or updated rows. | +| Deletion vectors | `deletion-vectors.enabled = true` | Deleted positions are marked in deletion-vector files. Updates mark old row versions as deleted and write new row versions. | + +Deletion vectors avoid rewriting an entire data file just to remove some rows. Readers apply the deletion information +when scanning the data. They do not remove the need to find matching rows or write updated values. + +![Deleting B produces the same visible rows A, C, and D. Copy on write replaces the affected file; deletion vectors retain it and mark B's position as deleted.](/img/append-row-level-storage.svg) + +The diagram shows a delete affecting part of one file. A delete that can remove a whole partition may instead use a +metadata-only operation. Replaced files can remain available to older snapshots until snapshot expiration makes them +eligible for cleanup. + +For example, create a Spark table with deletion vectors: + +```sql +CREATE TABLE mutable_events ( + event_id BIGINT, + payload STRING +) USING paimon +TBLPROPERTIES ( + 'deletion-vectors.enabled' = 'true' +); +``` + +Bucketed append tables with incremental clustering cannot enable deletion vectors. Check +[clustering requirements](./incremental-clustering#requirements) before combining these features. + +## Track Row Identity + +Enable [row tracking](./row-tracking) at table creation when you need a hidden row ID and a row version across updates +and ordinary compaction. Row tracking is supported only for unaware-bucket append tables (`bucket = -1`). It is separate +from deletion vectors and does not need to be enabled for the basic statements above. + +Neither feature turns streaming reads into a complete change feed for these mutations. See +[streaming read behavior](./streaming#overwrite-commits) when a table also has downstream streaming consumers. + +These examples cover regular append tables through Spark SQL. For the separate Data Evolution storage model and its +supported operations, see [Data Evolution](../multimodal-table/data-evolution). diff --git a/docs/docs/append-table/row-tracking.md b/docs/docs/append-table/row-tracking.md index a6737becb1f1..d3661ec78fcc 100644 --- a/docs/docs/append-table/row-tracking.md +++ b/docs/docs/append-table/row-tracking.md @@ -1,6 +1,6 @@ --- title: "Row Tracking" -sidebar_position: 5 +sidebar_position: 6 --- -# Row tracking +# Row Tracking -Row tracking allows Paimon to track row-level tracking in a Paimon append table. Once enabled on a Paimon table, two more hidden columns will be added to the table schema: -- `_ROW_ID`: BIGINT, this is a unique identifier for each row in the table. It is used to track the update of the row and can be used to identify the row in case of update, merge into or delete. -- `_SEQUENCE_NUMBER`: BIGINT, this is field indicates which `version` of this record is. It actually is the snapshot-id of the snapshot that this row belongs to. It is used to track the update of the row version. +Row tracking adds two hidden metadata columns to an append table. They distinguish a row's identity from the snapshot +in which its current version was written. -Hidden columns follows the following rules: -- Whenever we read from one table with row tracking enabled, the `_ROW_ID` and `_SEQUENCE_NUMBER` will be `NOT NULL`. -- If we append records to row-tracking table in the first time, we don't actually write them to the data file, they are lazy assigned by committer. -- If one row moved from one file to another file for **any reason**, the `_ROW_ID` column should be copied to the target file. The `_SEQUENCE_NUMBER` field should be set to `NULL` if the record is changed, otherwise, copy it too. -- Whenever we read from a row-tracking table, we firstly read `_ROW_ID` and `_SEQUENCE_NUMBER` from the data file, then we read the value columns from the data file. If they found `NULL`, we read from `DataFileMeta` to fall back to the lazy assigned values. Anyway, it has no way to be `NULL`. +| Column | Type | Meaning | +| --- | --- | --- | +| `_ROW_ID` | `BIGINT` | Paimon's identifier for a row, preserved across updates and ordinary compaction. | +| `_SEQUENCE_NUMBER` | `BIGINT` | The snapshot ID assigned when this version of the row was inserted or updated. Unchanged rows retain their version during ordinary compaction. | + +These fields are managed by Paimon and are non-null when read. A row version is not the ID of every snapshot that +contains the row: many later snapshots can still contain an unchanged row with an older sequence number. + +:::note Experimental + +Row tracking is experimental. Enable it when creating an unaware-bucket append table (`bucket = -1`, with no primary +key or bucket key). `row-tracking.enabled` is immutable and cannot be enabled later with `ALTER TABLE`. + +::: + +## Enable Row Tracking + +For example, create a partitioned table in Flink SQL: -To enable row-tracking, you must config `row-tracking.enabled` to `true` in the table options when creating an append table. -Consider an example via Flink SQL: ```sql CREATE TABLE part_t ( - f0 INT, - f1 STRING, + id INT, + data STRING, dt STRING -) PARTITIONED BY (dt) -WITH ('row-tracking.enabled' = 'true'); +) PARTITIONED BY (dt) WITH ( + 'bucket' = '-1', + 'row-tracking.enabled' = 'true' +); ``` -Notice that: -- Row tracking is only supported for unaware append tables, not for primary key tables. Which means you can't define `bucket` and `bucket-key` for the table. -- Only spark support update, merge into and delete operations on row-tracking tables, Flink SQL does not support these operations yet. -- This function is experimental, this line will be removed after being stable. -After creating a row-tracking table, you can insert data into it as usual. The `_ROW_ID` and `_SEQUENCE_NUMBER` columns will be automatically managed by Paimon. -```sql -CREATE TABLE t (id INT, data STRING) TBLPROPERTIES ('row-tracking.enabled' = 'true'); -INSERT INTO t VALUES (11, 'a'), (22, 'b') -``` +Insert data as usual; do not add the hidden columns to the user-defined schema. The following walkthrough uses +Spark SQL for both querying the metadata columns and performing row-level changes on a regular append table. + +## Follow a Row Through Changes + +![An update assigns a new row version while retaining the row ID. Ordinary compaction preserves both values.](/img/append-row-tracking.svg) + +### Insert and Read + +Create a separate, unpartitioned table in a Paimon Spark catalog: -You can select the row tracking meta column with the following sql in spark: ```sql -SELECT id, data, _ROW_ID, _SEQUENCE_NUMBER FROM t; +CREATE TABLE t (id INT, data STRING) USING paimon +TBLPROPERTIES ('row-tracking.enabled' = 'true'); + +INSERT INTO t VALUES (11, 'a'), (22, 'b'); +SELECT id, data, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id; ``` -You will get the following result: + +The results below illustrate an initially empty table with one commit per write statement and no intervening commits. +Row-ID assignment can depend on file and write parallelism; do not rely on a business key receiving a particular ID. + ```text +---+----+-------+----------------+ | id|data|_ROW_ID|_SEQUENCE_NUMBER| @@ -69,57 +88,85 @@ You will get the following result: +---+----+-------+----------------+ ``` -Then you can update and query the table again: +### Update + ```sql -UPDATE t SET data = 'new-data-update' WHERE id = 11; --- Alternatively, update using the hidden row id `_ROW_ID` -UPDATE t SET data = 'new-data-update' WHERE _ROW_ID = 0; -SELECT id, data, _ROW_ID, _SEQUENCE_NUMBER FROM t; +UPDATE t SET data = 'a2' WHERE id = 11; +SELECT id, data, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id; ``` -You will get: +The changed row retains its ID and receives a new sequence number. The untouched row keeps both values: + ```text -+---+---------------+-------+----------------+ -| id| data|_ROW_ID|_SEQUENCE_NUMBER| -+---+---------------+-------+----------------+ -| 22| b| 1| 1| -| 11|new-data-update| 0| 2| -+---+---------------+-------+----------------+ ++---+----+-------+----------------+ +| id|data|_ROW_ID|_SEQUENCE_NUMBER| ++---+----+-------+----------------+ +| 11| a2| 0| 2| +| 22| b| 1| 1| ++---+----+-------+----------------+ ``` -You can also merge into the table, suppose you have a source table `s` that contains (22, 'new-data-merge') and (33, 'c'): +You can alternatively match an update with `WHERE _ROW_ID = 0`, using the ID returned by a previous query. Run either +form once if you are following the illustrated sequence numbers. + +The sequence number records a write version, not a comparison of the old and new field values. A matching `UPDATE` +can assign a new sequence number even when the assigned value is the same, such as `UPDATE t SET data = data WHERE id = 11`. + +### Merge + ```sql -MERGE INTO t USING s -ON t.id = s.id +CREATE TEMPORARY VIEW s AS +SELECT * FROM VALUES (22, 'b2'), (33, 'c') AS source(id, data); + +MERGE INTO t USING s ON t.id = s.id WHEN MATCHED THEN UPDATE SET t.data = s.data WHEN NOT MATCHED THEN INSERT *; + +SELECT id, data, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id; ``` -You will get: +The updated row retains its ID; the inserted row receives a new ID. Both versions come from the merge commit: + ```text -+---+---------------+-------+----------------+ -| id| data|_ROW_ID|_SEQUENCE_NUMBER| -+---+---------------+-------+----------------+ -| 11|new-data-update| 0| 2| -| 22| new-data-merge| 1| 3| -| 33| c| 2| 3| -+---+---------------+-------+----------------+ ++---+----+-------+----------------+ +| id|data|_ROW_ID|_SEQUENCE_NUMBER| ++---+----+-------+----------------+ +| 11| a2| 0| 2| +| 22| b2| 1| 3| +| 33| c| 2| 3| ++---+----+-------+----------------+ ``` -You can also delete from the table: +### Delete ```sql DELETE FROM t WHERE id = 11; --- Alternatively, delete using the hidden row id `_ROW_ID` -DELETE FROM t WHERE _ROW_ID = 0; +SELECT id, data, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id; ``` -You will get: +The deleted row is no longer visible. The remaining rows retain their identity and version: + ```text -+---+---------------+-------+----------------+ -| id| data|_ROW_ID|_SEQUENCE_NUMBER| -+---+---------------+-------+----------------+ -| 22| new-data-merge| 1| 3| -| 33| c| 2| 3| -+---+---------------+-------+----------------+ ++---+----+-------+----------------+ +| id|data|_ROW_ID|_SEQUENCE_NUMBER| ++---+----+-------+----------------+ +| 22| b2| 1| 3| +| 33| c| 2| 3| ++---+----+-------+----------------+ ``` + +You can also delete by a previously queried `_ROW_ID`. These metadata columns do not turn the table into a primary key +table and do not deduplicate inserted business keys. + +## How Metadata Is Stored + +For newly appended rows, Paimon can assign IDs and sequence numbers lazily during commit using file metadata rather +than writing the hidden values into every row. Readers use stored row metadata when available and fall back to the +file metadata when a hidden value is absent. + +When an ordinary rewrite moves a row to another data file, its row ID is carried forward. Rows copied without being +updated keep their sequence numbers; updated rows receive a new sequence number at commit. Ordinary compaction +therefore does not by itself change a row's version. + +For the separate Data Evolution storage model, including maintenance that can reassign physical row IDs, see +[Data Evolution](../multimodal-table/data-evolution). diff --git a/docs/docs/append-table/streaming.mdx b/docs/docs/append-table/streaming.mdx new file mode 100644 index 000000000000..e740f05b055a --- /dev/null +++ b/docs/docs/append-table/streaming.mdx @@ -0,0 +1,191 @@ +--- +title: "Streaming" +sidebar_position: 1 +--- + + + +# Streaming + +Flink can continuously write to and read from an append table. New data becomes visible after a snapshot is committed; +end-to-end latency depends on checkpointing, commit time, and the reader's discovery interval. + +This page uses the default unaware-bucket layout (`bucket = -1`). For ordering within a fixed bucket, see +[Bucketed streaming](./bucketed#bucketed-streaming). + +## Write a Stream + +Run an `INSERT INTO` from an insert-only source in streaming mode. For example, after creating `my_table` from the +[overview](./) and a source table with the same columns: + +```sql +SET 'execution.runtime-mode' = 'streaming'; +SET 'execution.checkpointing.interval' = '1 min'; + +INSERT INTO my_table SELECT product_id, price, sales, dt FROM source_table; +``` + +Short checkpoint intervals can produce many small data files. Choose a compaction approach together with your +checkpoint interval and write parallelism. + +## Manage Small Files + +| Approach | When it runs | How to select it | +| --- | --- | --- | +| Pre-commit compaction | Merges newly written files from the same partition before they enter a snapshot. | Set `precommit-compact = true`; the default is `false`. | +| Background compaction in the ingestion job | Plans compaction from files already committed to the table. | Included in a normal Flink streaming sink for an unaware-bucket append table. | +| Dedicated compaction | Merges committed files in a separate job. | Set `write-only = true` on ingestion jobs and run a [dedicated compaction job](../maintenance/dedicated-compaction). | + +### Pre-Commit Compaction + +```sql +ALTER TABLE my_table SET ('precommit-compact' = 'true'); +``` + +This adds a coordinator and workers after the writer to merge newly created **data files**. The compaction runs before +commit, so its work contributes to the time needed to make those files visible. Configure the option before starting +the ingestion job. + +### Background Compaction + +For a normal unaware-bucket append table, the writer does not compact files itself. The Flink streaming sink adds a +compact coordinator and compact workers. The coordinator discovers committed small files, and the workers rewrite +selected files while forwarding ingestion committables to the committer. + +![Flink append sink with a writer, background compact coordinator and workers, and a snapshot committer.](/img/append-streaming-compaction.svg) + +The compaction work runs asynchronously, but it still consumes CPU, memory, and storage I/O. Size those resources for +both ingestion and compaction. Setting `write-only = true` removes the background compaction operators from ingestion; +`precommit-compact` is configured separately. + +:::note Incremental clustering + +For unaware-bucket tables with `clustering.incremental = true`, the sink does not add this background compaction path. +Schedule [incremental clustering](./incremental-clustering#run-incremental-clustering) to merge small files and maintain +the clustered layout. Bucketed tables use a different compaction path, described in that guide. + +::: + +### Dedicated Compaction + +To move background compaction out of the ingestion job, apply `write-only` to that job. For example, use this insert +instead of the one in [Write a stream](#write-a-stream): + +```sql +INSERT INTO my_table /*+ OPTIONS('write-only' = 'true') */ +SELECT product_id, price, sales, dt FROM source_table; +``` + +Then run a [dedicated compaction job](../maintenance/dedicated-compaction), or a scheduled clustering job if incremental +clustering is enabled. `write-only` also skips snapshot expiration in the ingestion job, so the dedicated maintenance +job must handle that work. It does not disable separately configured pre-commit compaction. + +## Read a Stream + +By default, a streaming read first reads the latest snapshot and then follows new records. To read only records +committed after the reader starts, use `scan.mode = latest`: + +```sql +SET 'execution.runtime-mode' = 'streaming'; + +-- Read the current snapshot, then follow new records. +SELECT * FROM my_table; + +-- Start from new records only. +SELECT * FROM my_table /*+ OPTIONS('scan.mode' = 'latest') */; +``` + +These are alternative queries, each starting its own read. For a specific starting snapshot or timestamp, see +[Flink streaming time travel](../flink/sql-query#streaming-time-travel). That guide covers `scan.snapshot-id`, +`scan.timestamp-millis`, and `scan.file-creation-time-millis` with their corresponding scan modes. + +Unaware-bucket tables do not guarantee row order. Bucketing can provide ordering within one partition and bucket; +it does not establish an order across the whole table. + +### Overwrite Commits + +Streaming reads ignore `INSERT OVERWRITE` commits by default. To include the added data files from an append-table +overwrite, enable `streaming-read-append-overwrite` on the read: + +```sql +SELECT * FROM my_table /*+ OPTIONS('streaming-read-append-overwrite' = 'true') */; +``` + +This reads the added rows; it does not emit retractions for the rows replaced by the overwrite. A downstream append +consumer can therefore see both the old rows and their replacements. The similarly named `streaming-read-overwrite` +option is for primary key tables and is not supported on append tables. + +Row tracking and deletion vectors do not make a regular append-table stream a complete row-level change feed. Use a +batch snapshot query to read the current table state after [row-level operations](./row-level-operations). + +## Event-Time Watermarks {#watermark-definition} + +You can declare a watermark when reading a Paimon table in Flink: + +```sql +CREATE TABLE events ( + user_id BIGINT, + product STRING, + order_time TIMESTAMP(3), + WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND +) WITH ( + 'bucket' = '-1' +); + +SELECT window_start, window_end, COUNT(user_id) +FROM TABLE(TUMBLE(TABLE events, DESCRIPTOR(order_time), INTERVAL '10' MINUTES)) +GROUP BY window_start, window_end; +``` + +Watermarks describe event-time progress; they do not sort records. For watermark alignment across sources, configure: + +| Option | Default | Purpose | +| --- | --- | --- | +| `scan.watermark.alignment.group` | Not set | Sources in the same group align their watermarks. | +| `scan.watermark.alignment.max-drift` | Not set | Maximum allowed drift before consumption is paused. | +| `scan.watermark.alignment.update-interval` | `1 s` | How often watermark alignment information is exchanged. | + +## Bounded Streaming {#bounded-stream} + +Set `scan.bounded.watermark` to end a streaming read when the reader encounters a snapshot whose stored watermark is +**greater than** the configured value. The value is a long integer, expressed in milliseconds for event-time watermarks. + +```sql +SELECT * FROM events +/*+ OPTIONS('scan.bounded.watermark' = '1799625600000') */; +``` + +The stopping condition uses the watermark committed by the **writer**. The upstream source must produce watermarks +and the ingestion job must propagate them into Paimon snapshots. Declaring a watermark only on the read side does not +populate snapshot watermarks. If snapshots have no watermark, or never advance beyond the threshold, this condition +will not end the read. + +This is a snapshot-level stopping condition. Add a `WHERE` predicate if the result must also satisfy a precise +row-level event-time cutoff. + +The starting snapshot and subsequent snapshots are handled differently: + +- If the selected startup mode reads an initial snapshot, that snapshot is read even if its watermark already exceeds + the bound; the stream then stops. +- When following subsequent snapshots, the reader stops **before** reading the first snapshot whose watermark exceeds + the bound. A watermark equal to the bound does not stop the read. + +Choose the [scan startup mode](#read-a-stream) together with the bound. The bound does not trim an initial snapshot +to the rows that existed at the corresponding event time. diff --git a/docs/docs/flink/sql-query.mdx b/docs/docs/flink/sql-query.mdx index ef2e3e9ae6b2..ec0ddf9f3d23 100644 --- a/docs/docs/flink/sql-query.mdx +++ b/docs/docs/flink/sql-query.mdx @@ -203,8 +203,9 @@ SELECT * FROM t /*+ OPTIONS('scan.file-creation-time-millis' = '1678883047356') ### Read Overwrite -Streaming reading will ignore the commits generated by `INSERT OVERWRITE` by default. If you want to read the -commits of `OVERWRITE`, you can configure `streaming-read-overwrite`. +Streaming reads ignore `INSERT OVERWRITE` commits by default. For primary key tables, enable `streaming-read-overwrite` +to read the changes. For append tables, use `streaming-read-append-overwrite` to read the added rows without retractions +for replaced rows; see [Append-table overwrite commits](../append-table/streaming#overwrite-commits). ## Read Parallelism diff --git a/docs/sidebars.js b/docs/sidebars.js index 086f4dc53fb3..17e29c1557b6 100644 --- a/docs/sidebars.js +++ b/docs/sidebars.js @@ -65,8 +65,11 @@ const sidebars = { "id": "append-table/index" }, "items": [ - "append-table/incremental-clustering", + "append-table/streaming", + "append-table/query-performance", "append-table/bucketed", + "append-table/incremental-clustering", + "append-table/row-level-operations", "append-table/row-tracking" ] }, diff --git a/docs/static/img/append-bucket-order.svg b/docs/static/img/append-bucket-order.svg new file mode 100644 index 000000000000..8f95d7e93718 --- /dev/null +++ b/docs/static/img/append-bucket-order.svg @@ -0,0 +1,59 @@ + + +Bucket-scoped streaming order +Within one partition, bucket 0 and bucket 1 each contain commits 1, 2, and 3. The reader for each bucket follows its append order. Independent readers have no ordering guarantee relative to one another. + + + +Append order is local to a bucket +One partition · bucket-append-ordered = true · earlier appends read first + +Bucket 0 + +Commit 1 +A, B + + +Commit 2 +C, D + + +Commit 3 +E, F + +Reader 1independent + +Bucket 1 + +Commit 1 +G, H + + +Commit 2 +I, J + + +Commit 3 +K, L + +Reader 2independent +No cross-bucket order: records from the two readers can interleave downstream. + + diff --git a/docs/static/img/append-file-pruning.svg b/docs/static/img/append-file-pruning.svg new file mode 100644 index 000000000000..07f7fe96dc4f --- /dev/null +++ b/docs/static/img/append-file-pruning.svg @@ -0,0 +1,63 @@ + + +File statistics prune more files after clustering +Before clustering, files A, B, and C contain values 10,120,310; 50,150,340; and 80,190,390. All three min/max ranges overlap the filter 100 through 200. After sorting the same nine values, file D contains 10,50,80 and file F contains 310,340,390, so both can be skipped. Only file E, containing 120,150,190, must be read. + + + +Clustering makes file ranges more selective +Same nine product IDs · predicate: product_id BETWEEN 100 AND 200 +BEFORE · read all three candidate files + +File A +10, 120, 310 +Min/max: 10–310 +READ · range overlaps + +File B +50, 150, 340 +Min/max: 50–340 +READ · range overlaps + +File C +80, 190, 390 +Min/max: 80–390 +READ · range overlaps + +cluster by product_id +AFTER · read one candidate file + +File D +10, 50, 80 +Min/max: 10–80 +SKIP · range outside filter + +File E +120, 150, 190 +Min/max: 120–190 +READ · range overlaps + +File F +310, 340, 390 +Min/max: 310–390 +SKIP · range outside filter +The reader still evaluates the predicate on rows in the selected files. + + diff --git a/docs/static/img/append-incremental-clustering.svg b/docs/static/img/append-incremental-clustering.svg new file mode 100644 index 000000000000..dbf9414b90f8 --- /dev/null +++ b/docs/static/img/append-incremental-clustering.svg @@ -0,0 +1,60 @@ + + +Incremental clustering selects a subset of files +An illustrative run selects new files A, B, and C at level 0 together with an eligible existing level 1 run. It sorts and merges that set into files D and E. An unselected higher-level run is reused without rewriting. Actual selections and output levels depend on the planner. + + + +Cluster the selected files; keep the rest +Illustrative incremental run within one partition +BEFORE +AFTER + +Selected for this run +Level 0 · new files + +A + +B + +C + +Level 1 · existing run +Sort + mergeclustering columns + + + +New clustered run + +File D + +File E +Target-sized output files + +Higher-level run +Not selected + +No rewrite + +Same higher-level run +Reused in the new snapshot +A full job can also skip already-clustered units. Output levels depend on the planner. + + diff --git a/docs/static/img/append-row-level-storage.svg b/docs/static/img/append-row-level-storage.svg new file mode 100644 index 000000000000..fe431fa0d53c --- /dev/null +++ b/docs/static/img/append-row-level-storage.svg @@ -0,0 +1,73 @@ + + +Copy on write and deletion vectors produce the same visible delete result +Original file F1 has values A, B, C, and D at positions 0, 1, 2, and 3. Deleting B with copy on write creates file F2 with A, C, and D and replaces the current reference to F1. With deletion vectors, F1 stays unchanged and a separate vector marks position 1 deleted. Both current reads return A, C, and D. Older snapshots can still reference replaced files. + + + +Deleting one row: two physical layouts +Delete value B · regular append table · current snapshot shown below + +Original file F1 +File positions: + +A +0 + +B +1 + +C +2 + +D +3 + + + + +Copy on write +Deletion vectors +Write replacement file F2 +Keep data file F1 unchanged + +A + +C + +D + +A + +B + +C + +D +The current snapshot references F2 +in place of F1. + +Deletion vector: F1, position 1 +The reader excludes the marked position. + +Visible result in both cases: A, C, D +Replaced files can remain referenced by older snapshots until expiration. + + diff --git a/docs/static/img/append-row-tracking.svg b/docs/static/img/append-row-tracking.svg new file mode 100644 index 000000000000..6a2a377aba20 --- /dev/null +++ b/docs/static/img/append-row-tracking.svg @@ -0,0 +1,51 @@ + + +Row ID stays stable while row version changes on update +A row inserted in snapshot 1 has data a, row ID 0, and sequence number 1. Updating it in snapshot 2 changes data to a2 and sequence number to 2, retaining row ID 0. Ordinary compaction in snapshot 3 retains both row ID 0 and sequence number 2. Deleting the row removes it from the current result. + + + +Row identity and row version +One row in a regular row-tracking table · illustrative snapshot IDs + +INSERT +Snapshot 1 +data = a +_ROW_ID = 0 +_SEQUENCE_NUMBER = 1 + +UPDATE +Snapshot 2 +data = a2 +_ROW_ID = 0 +_SEQUENCE_NUMBER = 2 + +COMPACT +Snapshot 3 +data = a2 +_ROW_ID = 0 +_SEQUENCE_NUMBER = 2 + + +Same ID across all three snapshots. +The update changes the row version; ordinary compaction preserves it. +A later DELETE removes the row from the current result. + + diff --git a/docs/static/img/append-streaming-compaction.svg b/docs/static/img/append-streaming-compaction.svg new file mode 100644 index 000000000000..cb234e247f44 --- /dev/null +++ b/docs/static/img/append-streaming-compaction.svg @@ -0,0 +1,48 @@ + + +Flink streaming compaction +Input reaches the writer without a bucket-key shuffle. Committables pass through the compact coordinator and workers to the committer. The coordinator discovers small files from committed snapshots, and workers compact asynchronously. Setting write-only to true removes the background compaction operators. + + + +Flink streaming compaction +Unaware-bucket append · background compaction enabled + +Background compaction operators +Inputinsert-only +Writernew files +Coordinatorparallelism = 1 +Workersasync rewrite +Committersnapshots + + + + +No bucket-key shuffle + +Table snapshots and data files +Committed files are candidates for later compaction + +discover files + +commit +write-only = true removes the background operators; pre-commit compaction is separate. + + diff --git a/docs/static/img/for-queue.png b/docs/static/img/for-queue.png deleted file mode 100644 index 5e453b7c1fbd..000000000000 Binary files a/docs/static/img/for-queue.png and /dev/null differ diff --git a/docs/static/img/unaware-bucket-topo.png b/docs/static/img/unaware-bucket-topo.png deleted file mode 100644 index f530fc4a225c..000000000000 Binary files a/docs/static/img/unaware-bucket-topo.png and /dev/null differ